Tools¶
Tools are executable operations — anything that takes arguments and
returns structured output. Tag a Litestar route handler with
mcp_tool="<tool_name>" and the plugin publishes it via tools/list
and tools/call. The task-manager demo registers five tool handlers
covering the full CRUD lifecycle:
docs/examples/task_manager/main.py - register_tools¶def register_tools(store: "dict[int, Task]") -> "list[Any]":
"""Return MCP tool handlers (task CRUD) bound to ``store``."""
# start-example
@get("/tasks", mcp_tool="list_tasks")
async def list_tasks(completed: "bool | None" = None) -> "list[Task]":
"""List all tasks, optionally filtered by completion status."""
if completed is None:
return list(store.values())
return [task for task in store.values() if task.completed == completed]
@get("/tasks/{task_id:int}", mcp_tool="get_task")
async def get_task(task_id: "int") -> "Task":
"""Get a specific task by ID."""
if task_id not in store:
raise NotFoundException(detail=f"Task {task_id} not found")
return store[task_id]
@post("/tasks", status_code=HTTP_201_CREATED, mcp_tool="create_task")
async def create_task(data: "CreateTaskRequest") -> "Task":
"""Create a new task."""
new_id = max(store.keys(), default=0) + 1
new_task = Task(id=new_id, title=data.title, description=data.description, completed=False)
store[new_id] = new_task
return new_task
@post("/tasks/{task_id:int}/complete", mcp_tool="complete_task")
async def complete_task(task_id: "int") -> "Task":
"""Mark a task as completed."""
if task_id not in store:
raise NotFoundException(detail=f"Task {task_id} not found")
store[task_id].completed = True
return store[task_id]
@delete("/tasks/{task_id:int}", mcp_tool="delete_task")
async def delete_task(task_id: "int") -> "None":
"""Delete a task by ID."""
if task_id not in store:
raise NotFoundException(detail=f"Task {task_id} not found")
del store[task_id]
# end-example
return [list_tasks, get_task, create_task, complete_task, delete_task]
The handlers themselves are ordinary Litestar @get / @post /
@delete callables — the only extra is the mcp_tool kwarg. Each
tool is discoverable via tools/list and invocable via
tools/call.
Tool arguments are validated against the handler's parsed_fn_signature
before dispatch — the same model Litestar uses for ordinary HTTP request
parsing. Missing required arguments surface as JSON-RPC
INVALID_PARAMS (-32602). Annotated[T, Parameter(...)] query
arguments are unwrapped and their Parameter constraints
(ge / le / min_length / pattern / …) flow through into
the advertised inputSchema.
Explicit Input Schemas¶
Use mcp_tool() with input_schema= when the generated
Litestar schema cannot express the exact client-facing JSON Schema 2020-12
contract. The explicit schema replaces inference for discovery and bridge
header generation, so keep it aligned with the handler's actual validation:
docs/examples/snippets/tool_explicit_input_schema.py¶@mcp_tool(
"search_regions",
input_schema={
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {"region": {"type": "string"}},
"required": ["region"],
"additionalProperties": False,
},
)
@get("/regions")
async def search_regions(region: str) -> list[str]:
return [region]
Task Input Before Execution¶
For a task-capable tool that must complete an MRTR input round before task
creation, combine task_support with task_input_before_start=True:
docs/examples/snippets/tool_task_input_before_start.py¶@mcp_tool(
"publish_report",
task_support="required",
task_input_before_start=True,
)
@post("/reports")
async def publish_report() -> MCPInputRequiredResult | dict[str, str]:
context = get_mcp_request_context()
if not context.input_responses:
return MCPInputRequiredResult(
input_requests={
"approval": {
"method": "elicitation/create",
"params": {"message": "Publish this report?"},
}
},
request_state="integrity-protected-state",
)
return {"status": "published"}
The first response is synchronous input_required and has no task ID. A
retry carrying inputResponses may create the task. Integrity-protect
authorization-sensitive requestState and bind it to the authenticated
principal, original arguments, and an expiry.
JSON-RPC Round-Trip¶
Clients drive tools with independent tools/list and tools/call POST
requests. Every request includes the 2026-07-28 metadata envelope and matching
HTTP routing headers:
# List every tool marked in the application
curl -sS -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-H "MCP-Protocol-Version: 2026-07-28" \
-H "Mcp-Method: tools/list" \
-d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}}}'
# Execute a specific tool (task-manager demo)
curl -sS -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-H "MCP-Protocol-Version: 2026-07-28" \
-H "Mcp-Method: tools/call" \
-H "Mcp-Name: list_tasks" \
-d '{"jsonrpc":"2.0","id":3,"method":"tools/call",
"params":{"name":"list_tasks","arguments":{},
"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28",
"io.modelcontextprotocol/clientCapabilities":{}}}}'
Successful responses carry the handler's return value inside the standard JSON-RPC envelope.
Binary and Mixed Content Results¶
For ordinary JSON-like return values, the plugin keeps the existing behavior: the value is serialized into a text content block. When a tool needs to return MCP content blocks directly, use the public helper types:
docs/examples/snippets/tool_binary_content.py¶@get("/reports/latest", mcp_tool="generate_report")
async def generate_report() -> "MCPToolResult":
return MCPToolResult(
content=[
{"type": "text", "text": "Report generated."},
MCPResourceLink(
name="report.pdf",
uri="litestar://latest_report",
mime_type="application/pdf",
),
],
structured_content={"reportId": "latest"},
)
@get("/reports/inline", mcp_tool="download_report")
async def download_report() -> "MCPBlobResource":
payload = b"...pdf bytes..."
return MCPBlobResource(
uri="memory://reports/latest.pdf",
data=payload,
mime_type="application/pdf",
)
@get(
"/reports/resource",
mcp_resource="latest_report",
mcp_resource_mime_type="application/pdf",
)
async def latest_report() -> "Response[bytes]":
return Response(content=b"...pdf bytes...", media_type="application/pdf")
app = Litestar(route_handlers=[generate_report, download_report, latest_report], plugins=[LitestarMCP()])
Use MCPResourceLink when the client should fetch bytes
later with resources/read. Use MCPBlobResource only
when the bytes need to be embedded immediately in the JSON-RPC response. A
handler returning raw bytes directly is treated like an ordinary handler
return value, not as an implicit blob.
Error Contract¶
Tool errors are reported differently from the other primitives. A handler
that raises or returns an error response is surfaced inside the tool
result with isError: true (and the detail in content), not as a
JSON-RPC error object — this lets the model see and react to the
failure. Only protocol-level problems (an unknown tool name, malformed
params) use a JSON-RPC error with INVALID_PARAMS (-32602).
See Prompts and Resources for how those primitives map the
handler's HTTP status onto JSON-RPC codes.