What it is
MCP (Model Context Protocol) lets AI agents call your app's Python functions the
same way htmx calls your routes. Register a function with@app.tool()and it
serves two callers from one codebase: your HTTP handlers (which call it directly)
and MCP clients over JSON-RPC.
Reach for this when you want an LLM agent to act on the same data your HTML UI exposes.
from chirp import App
app = App()
@app.tool("search_inventory", description="Search inventory by keyword")
async def search_inventory(query: str, limit: int = 10) -> list[dict]:
return await db.search(query, limit=limit)
That function is now callable from:
- Your HTTP handlers — call it directly, like any other function.
- MCP clients — over JSON-RPC at
/mcp.
Registering tools
Use the@app.tool()decorator during setup. The first argument is the tool
name;descriptionis sent to MCP clients so agents know what each tool does.
@app.tool("add_note", description="Add a note with an optional tag.")
def add_note(text: str, tag: str | None = None) -> dict:
note = {"id": next_id(), "text": text, "tag": tag}
store.append(note)
return note
@app.tool("list_notes", description="List all notes.")
def list_notes() -> list[dict]:
return list(store)
Both sync and async handlers work.
Chirp generates JSON Schema from your type annotations, so MCP clients get a
typed parameter list for free. Parameters namedrequestare excluded (the same
convention as route handlers).
Type-to-schema mapping
The schema is built at freeze time from each parameter's annotation:
str→"string",int→"integer",float→"number",bool→"boolean"list[str]→"array"with"items": {"type": "string"}(alsolist[int],list[float])X | None→ optional parameter (unwrapped toX, left out ofrequired)- Parameters with a default value are optional
- Parameters named
request(or annotatedRequest) are excluded - Unannotated parameters default to
"string"
The MCP endpoint
When at least one tool is registered, Chirp mounts a JSON-RPC endpoint at/mcp.
It speaks MCP protocol version2024-11-05 with the toolscapability. The
endpoint handshakes in three sequential calls — initialize, list, then call:
- 1
Initialize
Negotiate capabilities. The server replies with its
protocolVersion.curl -X POST http://localhost:8000/mcp \ -H 'Content-Type: application/json' \ -d '{"jsonrpc":"2.0","method":"initialize","id":1,"params":{}}' - 2
List tools
Fetch the registered tools and their input schemas.
curl -X POST http://localhost:8000/mcp \ -H 'Content-Type: application/json' \ -d '{"jsonrpc":"2.0","method":"tools/list","id":2,"params":{}}' - 3
Call a tool
Dispatch a tool by name with arguments.
curl -X POST http://localhost:8000/mcp \ -H 'Content-Type: application/json' \ -d '{"jsonrpc":"2.0","method":"tools/call","id":3,"params":{"name":"add_note","arguments":{"text":"Hello"}}}'
Real-time tool activity
Every successful tool call emits aToolCallEvent through app.tool_events.
Subscribe from an SSE route to build a live agent-activity dashboard.
from chirp import EventStream, Fragment
@app.route("/activity/feed", referenced=True)
def activity_feed():
async def stream():
async for event in app.tool_events.subscribe():
yield Fragment("dashboard.html", "activity_row", event=event)
return EventStream(stream())
EventStreamis one of Chirp's return types; for
the wire format and connection lifecycle see
Server-Sent Events.
EachToolCallEventis a frozen dataclass with:
tool_name— which tool was calledarguments— the arguments passedresult— what it returnedtimestamp— when it was called (epoch seconds)call_id— a unique 12-character hex identifier
Render an event in a template block like any other context value:
{% block activity_row %}
<tr>
<td><code>{{ event.tool_name }}</code></td>
<td>{{ event.arguments | format_args }}</td>
<td>{{ event.call_id[:8] }}</td>
</tr>
{% endblock %}
Inspecting the registry
After the app is frozen (first request orapp.run()), app.toolsreturns the
frozenToolRegistry. It is read-only at runtime; accessing it before freeze
raisesRuntimeError.
for tool_info in app.tools.list_tools():
print(f"{tool_info['name']}: {tool_info['description']}")
# Look up a specific tool
tool = app.tools.get("add_note")
if tool is not None:
print(tool.schema)
Thread safety
The tools system is built for Python 3.14 free-threading:
ToolDefis a frozen dataclass — immutable, safe to share across threads.ToolRegistryis built once at freeze time and never mutated.ToolEventBusguards its subscriber set with athreading.Lock.- Each subscriber gets its own
asyncio.Queue, so there is no shared mutable state on the broadcast path.
For the framework-wide model see the free-threading thread-safety model.
The shipping example
The runnable demo registers three tools, serves a notes UI, and streams tool calls into a live activity feed. The tool definitions:
@app.tool("add_note", description="Add a note with an optional tag.")
def add_note(text: str, tag: str | None = None) -> dict:
global _next_id
with _lock:
note = {"id": _next_id, "text": text, "tag": tag}
_next_id += 1
_notes.append(note)
return note
@app.tool("list_notes", description="List all notes.")
def list_notes() -> list[dict]:
with _lock:
return list(_notes)
@app.tool("search_notes", description="Search notes by text substring.")
def search_notes(query: str) -> list[dict]:
with _lock:
q = query.lower()
return [n for n in _notes if q in n["text"].lower()]
Source: examples/standalone/tools/app.py.
Run the full example withpython app.py, open it in a browser, then call a
tool withcurland watch the activity feed update in real time.