duck.contrib.mcp

Minimal MCP (Model Context Protocol) server support for Duck, mirroring the WebSocketView pattern. Speaks JSON-RPC 2.0 over a single HTTP endpoint (MCP “Streamable HTTP” transport): POST for requests, GET for an optional server-push stream.

By default each request gets one JSON response. SSE is opt-in per view via sse = True and only used when the client also asks for it via Accept: text/event-stream - in that case the response is streamed over the raw socket as Server-Sent Events, and tools/resources/prompts can push progress notifications mid-call via self.notify(...).

Subpackages

• duck.contrib.mcp.capabilities
  • duck.contrib.mcp.capabilities.accessor
  • duck.contrib.mcp.capabilities.defaults

Submodules

• duck.contrib.mcp.auth
• duck.contrib.mcp.codes
• duck.contrib.mcp.decorators
• duck.contrib.mcp.exceptions
• duck.contrib.mcp.session
• duck.contrib.mcp.testserver

Package Contents

Classes

MCPView

Base class for an MCP server endpoint.

API

class duck.contrib.mcp.MCPView(*args, **kwargs)

Bases: duck.views.View

Base class for an MCP server endpoint.

Subclass it, decorate methods with @tool / @resource / @prompt, and wire the class up with path() like any other view:

class SomeMCPServer(MCPView):
    name = "duck-mcp-server"
    version = "1.0.0"

    @tool(description="Add two numbers")
    async def add(self, a: int, b: int) -> int:
        return a + b

urlpatterns = [
    path('/mcp', SomeMCPServer, name="mcp_endpoint"),
]

For anything beyond the built-in methods, register a Capability instead of adding one-off handlers by hand. Every method whose prefix matches the registered name is routed there automatically:

class UsageCapability(Capability):
    def setup(self):
        self.handlers = {"report": self.report}

    async def report(self, params):
        return {"calls": 42}

class SomeMCPServer(MCPView):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.register_capability("usage", UsageCapability(self))

A request for usage/report now dispatches to UsageCapability.report.

Initialization

Collect all @tool / @resource / @prompt decorated methods on this instance into lookup dicts.

SSE_CHUNK_SIZE

None

int: max bytes written to the socket per send call while streaming an SSE event, so large payloads don’t block the event loop in one write.

SSE_PING_INTERVAL

15

int: seconds between keep-alive pings on a standalone GET SSE stream, so intermediaries don’t time out an otherwise idle connection.

_session_id_header

‘Mcp-Session-Id’

The MCP session ID header. Defaults to ‘mcp-session-id’.

_session_queues

None

dict[str, asyncio.Queue]: process-wide registry mapping MCP session IDs to the queue backing that session’s standalone GET SSE stream, so notify() calls from any request - even one running in a completely different instance, e.g. a background job kicked off by a tool call - can reach a stream opened elsewhere. Class-level so it’s shared across all instances of this view.

allowed_origins

None

list[str] | None: hostnames to validate the Origin header against (DNS-rebinding protection, as recommended by the MCP spec for locally- bound servers). None (default) falls back to settings.ALLOWED_HOSTS, so there’s nothing to configure twice - pass an explicit list on a subclass to override. "*" in the list disables the check entirely, matching Django-style ALLOWED_HOSTS semantics.

async assign_session_if_new(method: str) → dict

Mint a session id on a successful ‘initialize’ call that didn’t already have one, firing on_session_create. Must run before any after_dispatch hook — the default save hook needs the id to exist first, or the session’s initial state is silently dropped.

Returns response headers carrying the new id (empty dict otherwise).

property attrs: Dict[str, Any]

Method for getting MCP view attributes, with stripped __ private attributes. This must be called after full object initiazation.

async authenticate() → duck.contrib.mcp.auth.AuthResult

Override to gate access to this MCP server. Must return an AuthResult - default is AuthResult(), i.e. allowed with no scope restriction. Pass scopes= to grant specific scopes, which authorize() then checks automatically against any scopes= declared on a @tool/@resource/@prompt:

async def authenticate(self):
    token = self.request.headers.get("Authorization", "").removeprefix("Bearer ")
    claims = verify_jwt(token)
    if claims is None:
        return AuthResult.deny()
    return AuthResult(scopes=claims["scopes"])

async authorize(scopes: list) → bool

Check whether the authenticated client has required scopes.

Called before a tool/resource/prompt with declared scopes= runs. Default checks those scopes against self.granted_scopes (set from the AuthResult returned by authenticate()); None there means unrestricted.

In the common case you never need to override this - just return the right scopes from authenticate(). Override only for non-scope-based logic (e.g. per-name rules).

cleanup_resources()

Ensure all capabilities are removed - ensuring cleanup on every capability.

Safe to delete session-scoped resources here.

empty_response() → duck.http.response.JsonResponse

Return an empty 200 OK JSON response.

async ensure_sse_event_sent(message: dict, event: str = 'message')

Tries writting single Server-Sent Event to the socket only if SSE has been initialized else the message is added to session queue.

error_response(rpc_id, code: Union[int, duck.contrib.mcp.codes.MCPErrorCode], message, status=200)

Build a JSON-RPC 2.0 error response envelope.

async finalize_initial_sse_response(response: duck.http.response.HttpResponse)

Finalizes the first/initial response that gets sent to client when initiating SSE.

… admonition:: Notes

Sets the CORS headers needed for browser-based clients (e.g. MCP Inspector’s “direct” mode) to read this response and its Mcp-Session-Id header.

get_handler(method: str)

Resolve a JSON-RPC method to a callable.

… admonition:: Notes

Checks any capability registered for the method’s namespace - the part before the ‘/’. e.g. “tools/call” falls back to whatever was registered under the “tools” namespace, dispatching “call”.

Returns:

A callable of shape handler(params), or None if nothing matches.

handle_preflight() → duck.http.response.HttpResponse

Answer a CORS preflight OPTIONS request with a 204 and no body - just the Access-Control-* headers the browser needs before it will send the actual POST/GET.

async handle_rpc() → Optional[duck.http.response.JsonResponse]

Validate origin, authenticate the request, parse it as JSON-RPC 2.0, and dispatch it to the matching MCP method handler.

async handle_session_delete()

Terminate an MCP session and clean up associated resources.

async handle_single_sse(rpc_id, method, handler, params) → Optional[duck.http.response.JsonResponse]

Handle a single JSON-RPC call by streaming its response back as Server-Sent Events on this same connection, instead of one JSON body.

… admonition:: Notes

Opens the SSE response immediately, then runs handler(params) with self._sse_initiated = True so any self.notify(...) made during the call - e.g. tool progress - writes straight to this socket as it happens.

The call’s own result (or error) is sent last as a single message event, then the stream closes. Scoped to this one request/response, unlike the persistent channel handle_sse_stream() opens.

Raises:

ExpectingNoResponse – Since the response has already been written directly to the socket.

async handle_sse_stream() → None

Open a standalone SSE stream for a bare GET request - the MCP spec’s optional server-push channel, separate from the per-call stream a POST gets via handle_single_sse().

… admonition:: Notes

Requires an Mcp-Session-Id header (issued on initialize) so notify() calls from other requests in the same session - e.g. a background job kicked off by a tool call - can be routed here via the shared _session_queues registry. Held open with periodic pings so intermediaries don’t close it as idle.

Raises:

ExpectingNoResponse – Since the response has already been written directly to the socket.

async handle_streamable_http(raw_body: str) → duck.http.response.JsonResponse

Parse a POST body as JSON-RPC 2.0 and dispatch it to the matching handler.

… admonition:: Notes

Replies with a single JSON body, unless the client asked for SSE and self.sse is on - then handle_single_sse() streams the response instead.

property has_session_id_header: bool

Checks whether the current request has session id header.

property mcp_registry: Dict[str, Callable]

Registry of MCP-decorated methods.

Maps method names to callables with an mcp_kind attribute. Any additional MCP metadata is stored directly on the callables by their decorators.

property message_queue

Returns the message Queue for sending messages to client.

name

‘duck-mcp-server’

async notify(method: str, params: dict = None)

Push a JSON-RPC notification (e.g. progress) to the client. If this request is itself streaming a response (inside handle_sse()), it’s sent directly on this socket. Otherwise, if this request’s session has a standalone GET stream open elsewhere (via handle_sse_stream()), it’s queued for delivery there. If neither applies, this is a no-op, so tools can call it unconditionally:

@tool(description="Process a big job")
async def process(self, job_id: str) -> str:
    await self.notify("notifications/progress", {"progress": 50})
    ...
    return "done"

property persistent_state: object

Return an object whose attributes persist for the lifetime of the session.

You can attach custom attributes to this object to store session-specific state without modifying the session itself.

protocol_version

‘2024-11-05’

register_capability(name: str, capability: duck.contrib.mcp.capabilities.Capability, alias: Optional[str] = None) → duck.contrib.mcp.capabilities.Capability

Register a capability under a namespace, routing every ‘{name}/*’ method to it.

Parameters:

• name – The namespace to claim, e.g. “tools” claims “tools/list”, “tools/call”, etc.

• capability – The Capability instance to route matching methods to.

• alias – Optional short name for direct access via self.capabilities.{alias}. If not provided, you can still access via self.capabilities.{name}.

Returns:

The same instance, for chaining.

Return type:

Capability

… rubric:: Example

self.register_capability("tools", ToolsCapability(self))

register_default_capabilities()

Registers the default capabilities e.g. InitializeCapability, ToolsCapability, etc.

register_default_hooks()

Registers the default hooks.

register_hook(hook_name: str, callback: callable) → None

Register a hook callback for extension points.

Parameters:

• hook_name – Name of the hook point (e.g., “before_dispatch”)

• callback – Async function to call at the hook point

Hook Points: - before_dispatch: (body, rpc_id, method, params) -> (handled, response) Called before dispatching a JSON-RPC method. If handled is True, returns response immediately.

- after_dispatch: (rpc_id, method, result) -> modified_result
    Called after a successful dispatch. Can modify the result.

- before_sse_send: (message) -> modified_message
    Called before sending an SSE event. Can modify the message.

- after_sse_send: (message) -> None
    Called after sending an SSE event.

- on_session_create: (session_id) -> None
    Called when a new session is created.

- on_session_delete: (session_id) -> None
    Called when a session is deleted.

async run() → Optional[duck.http.response.JsonResponse]

Entry point, required by the routing dispatch convention. Delegates to handle_rpc().

async run_hooks(hook_name: str, *args, **kwargs)

Run all registered hooks for a given hook point.

Returns:

(handled, response) - first hook that handles the request stops further hooks. For other hooks: the last returned value, or None.

Return type:

For before_dispatch

async send_sse_event(message: dict, event: str = 'message')

Write a single Server-Sent Event (event: + data: lines, blank line terminated) to the socket, in SSE_CHUNK_SIZE-byte pieces so a large payload doesn’t block the event loop with one giant write.

property session_id: Optional[None]

Return the session ID for the client.

property session_id_header: str

Returns the formatted session ID header.

property sock

Returns the connected socket.

sse

False

bool: Set True to allow this server to stream responses as Server-Sent Events. Only takes effect on requests that also send Accept: text/event-stream - plain requests still get a single JSON response either way.

strictly_async()

unregister_capability(name: str) → duck.contrib.mcp.capabilities.Capability

Remove a previously registered capability, running its cleanup() hook first.

unregister_hook(hook_name: str, callback: callable) → None

Remove a previously registered hook callback.

validate_origin() → bool

Check the request’s Origin header against allowed_origins. Missing Origin always passes - most non-browser MCP clients don’t send one - this only catches a present-but-mismatched Origin, which is the DNS-rebinding/malicious- webpage case it’s meant for.

version

‘0.1.0’

wants_sse() → bool

Whether the client’s Accept header asks for an SSE stream rather than a plain JSON response.