Skip to content

MCPServer

MCPServer

A pluggable MCP server backed by ServiceSpec registrations.

The server owns its tool and resource registries, an auth backend, and a session store — all instance state, no module-level singletons. Two parallel registration shapes are supported:

Imperative::

server = MCPServer(name="my-app")
server.register_service_tool(
    name="invoices.create",
    spec=ServiceSpec(service=create_invoice, input_serializer=InvoiceInput),
)
server.register_resource(
    name="invoice",
    uri_template="invoices://{pk}",
    selector=SelectorSpec(selector=get_invoice, output_serializer=InvoiceOutput),
)

Declarative::

@server.service_tool(name="invoices.create", input_serializer=InvoiceInput)
def create_invoice(*, data): ...

@server.resource(uri_template="invoices://{pk}", output_serializer=InvoiceOutput)
def get_invoice(*, pk): ...

Mount the URLs in your URL conf:

urlpatterns = [path("mcp/", include(server.urls))]

urls property 

urls: list[URLPattern]

Sync URL patterns. Suitable for any deployment (WSGI or ASGI).

Use :attr:async_urls instead when running under ASGI to get non-blocking dispatch for the I/O-bound handlers.

async_urls property 

async_urls: list[URLPattern]

Async URL patterns for ASGI deployments.

tools/call, resources/read, and prompts/get dispatch through async-native runners; sync collaborators (auth backend, session store, custom permissions) are bridged via :func:asgiref.sync.sync_to_async so a fully sync stack still works. Async-native backends are detected by signature and called directly.

register_service_tool 

register_service_tool(
    *,
    name: str,
    spec: ServiceSpec,
    description: str | None = None,
    title: str | None = None,
    output_format: OutputFormat | str = OutputFormat.JSON,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
) -> ToolBinding

Register a :class:ServiceSpec as an MCP mutation tool.

Mirrors :meth:register_resource's spec-only contract — the unit of registration is a ServiceSpec from djangorestframework-services. The dispatch pipeline runs input_serializer → run_service(atomic) → output_selector? → output_serializer, so this is the right surface for side-effecting operations (creates, updates, deletes, anything that wants transaction.atomic()).

For read-shaped operations (list/retrieve with optional filtering / ordering / pagination) use :meth:register_selector_tool instead — selectors return raw querysets and the tool layer owns the post-fetch pipeline.

register_selector_tool 

register_selector_tool(
    *,
    name: str,
    spec: SelectorSpec,
    description: str | None = None,
    title: str | None = None,
    input_serializer: type | None = None,
    output_format: OutputFormat | str = OutputFormat.JSON,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
    filter_set: Any | None = None,
    ordering_fields: list[str] | tuple[str, ...] | None = None,
    paginate: bool = False,
) -> SelectorToolBinding

Register a :class:SelectorSpec as an MCP read tool.

Read-shaped sibling of :meth:register_service_tool. The selector returns a raw, unscoped queryset; the tool layer owns the post-fetch pipeline:

.. code-block:: text

arguments → validate(merged inputSchema)
          → run_selector
          → FilterSet(data=...).qs    (if filter_set set)
          → order_by(...)             (if ordering_fields set)
          → paginate                  (if paginate=True)
          → output_serializer(many=True)
          → ToolResult

Each pipeline knob is optional. A selector tool with none of filter_set / ordering_fields / paginate set behaves like a plain RPC read against the selector — same effective contract as a service tool minus the side effects.

filter_set requires the [filter] extra (django-filter). The constructor surfaces a clear ImportError if you set it without the package installed.

register_resource 

register_resource(
    *,
    name: str,
    uri_template: str,
    selector: SelectorSpec,
    description: str | None = None,
    title: str | None = None,
    output_serializer: type | None = None,
    mime_type: str = "application/json",
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
) -> ResourceBinding

Register a :class:SelectorSpec as an MCP resource.

The unit of registration is a spec, mirroring :meth:register_service_tool's :class:ServiceSpec requirement. selector.selector is the callable dispatched at resources/read time; selector.output_serializer fills in when the caller didn't pass one explicitly (the explicit output_serializer= kwarg wins); selector.kwargs becomes the binding's per-request kwargs provider.

Bare callables are no longer accepted at this surface — wrap them in SelectorSpec(selector=fn), or use :meth:resource (the decorator form), which wraps the function automatically.

register_prompt 

register_prompt(
    *,
    name: str,
    render: Callable[..., Any],
    description: str | None = None,
    title: str | None = None,
    arguments: list[PromptArgument] | None = None,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
) -> PromptBinding

Register a render callable as an MCP prompt.

render receives the prompt arguments as kwargs (plus request and user if it declares them) and returns either a string, a list of strings, a list of :class:PromptMessage, or a coroutine yielding any of those — the dispatch layer normalises the result.

service_tool 

service_tool(
    *,
    name: str,
    spec: ServiceSpec | None = None,
    input_serializer: type | None = None,
    output_serializer: type[Serializer] | None = None,
    output_selector: Callable[..., Any] | None = None,
    atomic: bool = True,
    success_status: int | None = None,
    description: str | None = None,
    title: str | None = None,
    output_format: OutputFormat | str = OutputFormat.JSON,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator form of :meth:register_service_tool.

If spec is supplied it is used verbatim; otherwise a :class:ServiceSpec is constructed from the keyword arguments. The original function is returned unchanged so it remains callable from Python without going through the MCP transport.

selector_tool 

selector_tool(
    *,
    name: str,
    spec: SelectorSpec | None = None,
    input_serializer: type | None = None,
    output_serializer: type[Serializer] | None = None,
    description: str | None = None,
    title: str | None = None,
    output_format: OutputFormat | str = OutputFormat.JSON,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
    filter_set: Any | None = None,
    ordering_fields: list[str] | tuple[str, ...] | None = None,
    paginate: bool = False,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator form of :meth:register_selector_tool.

If spec is supplied it is used verbatim; otherwise a :class:SelectorSpec is constructed from the wrapped function and the keyword arguments. The original function is returned unchanged so it remains callable from Python without going through the MCP transport.

resource 

resource(
    *,
    uri_template: str,
    name: str | None = None,
    spec: SelectorSpec | None = None,
    description: str | None = None,
    title: str | None = None,
    output_serializer: type[Serializer] | None = None,
    mime_type: str = "application/json",
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator form: register the wrapped callable as a resource.

If spec is supplied it is used verbatim; otherwise a :class:SelectorSpec is constructed from the wrapped function and the keyword arguments. The original function is returned unchanged so it remains callable from Python without going through the MCP transport.

prompt 

prompt(
    *,
    name: str | None = None,
    description: str | None = None,
    title: str | None = None,
    arguments: list[PromptArgument] | None = None,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator form: register the wrapped callable as a prompt.

notify async 

notify(session_id: str, payload: Any) -> bool

Push a JSON-RPC payload to a session's open SSE stream.

Returns True if a subscriber was present, False if no client is currently connected. Most callers will fire-and-forget — a missed push is not generally an error, since clients can pull state via tools/call round-trips. The broker enforces single-subscriber semantics: re-subscribing replaces the old queue silently.

When a :class:SSEReplayBuffer is configured the payload is recorded before publishing so that:

  • The published frame carries an event ID the SSE generator emits on the wire (id: <id>\ndata: <payload>\n\n).
  • A subsequent reconnect with Last-Event-ID can drain the missed events from the buffer before resuming live mode.

Without a buffer the wire shape is unchanged (no id: lines) and resume is disabled.

Multi-process deployments need an out-of-process broker (e.g. Redis pub/sub) to fan out across worker processes; the in-process broker only sees its own worker.

MCPServiceView

MCPServiceView dataclass

Minimal :class:rest_framework_services.ServiceView adapter for MCP.

Per-spec kwargs providers on ServiceSpec / SelectorSpec (added in djangorestframework-services 0.6) are typed against :class:~rest_framework_services.ServiceView — a structural Protocol requiring request, kwargs, and action. The MCP transport doesn't go through DRF views, so we synthesise an instance that satisfies the Protocol and pass it into the provider.

  • request is the internal DRF Request the dispatch flow already builds (so providers can read request.user etc.).
  • kwargs is the URI-template variables on resource reads, or an empty dict on tool calls.
  • action is the binding name ("invoices.create", etc.) — gives providers a stable identifier without digging at view internals.