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,
    display_name: str | None = None,
    display_description: 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,
    include_structured_content: bool | None = None,
    include_output_schema: bool | None = None,
    argument_binding: ArgumentBinding = ArgumentBinding.DATA_ONLY,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
    spec_kwargs_provides: tuple[str, ...] = (),
) -> 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,
    display_name: str | None = None,
    display_description: 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,
    include_structured_content: bool | None = None,
    include_output_schema: bool | None = None,
    argument_binding: ArgumentBinding = ArgumentBinding.MERGE,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
    spec_kwargs_provides: tuple[str, ...] = (),
) -> 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.

The selector's shape (LIST vs RETRIEVE) is read from spec.kind — a required field on SelectorSpec in djangorestframework-services 0.13+. LIST runs the full post-fetch pipeline (filter_set / ordering_fields / paginate) and renders with many=True; RETRIEVE rejects those pipeline knobs at registration and renders the result with many=False.

register_chain_tool 

register_chain_tool(
    *,
    name: str,
    steps: list[ChainStep] | tuple[ChainStep, ...],
    description: str | None = None,
    title: str | None = None,
    display_name: str | None = None,
    display_description: str | None = None,
    input_serializer: type | None = None,
    atomic: bool = True,
    output_alias: str | None = None,
    output_all: bool = False,
    output_format: OutputFormat | str = OutputFormat.JSON,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
    include_structured_content: bool | None = None,
    include_output_schema: bool | None = None,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
) -> ChainToolBinding

Register an ordered sequence of specs as a single MCP tool.

Each :class:~rest_framework_mcp.registry.types.chain_step.ChainStep wraps a ServiceSpec (write) or SelectorSpec (read) and binds its result to an alias. A step's inputs callable reads the validated tool arguments (ctx.args) and any prior step's output (ctx[alias]) to build that step's call kwargs — so one tool call can express retrieve x → write y → write z with z derived from both x and y.

atomic=True (the default) runs the whole sequence inside one transaction.atomic(): any step raising a ServiceError / ServiceValidationError rolls back every prior write and the JSON-RPC error carries failedStep.

The advertised inputSchema is input_serializer when set, otherwise the first step's serializer (the first-step fallback). The response is the output_alias step's rendered output (default: the last step), or {alias: rendered} for every serializer-bearing step when output_all=True.

Each step's spec.permission_classes are AND-combined with the chain-level permissions and evaluated up front — a failing step permission blocks the whole chain before any step runs.

Chains deliberately do not run the selector post-fetch pipeline (filter / order / paginate); for that, expose the selector as its own :meth:register_selector_tool.

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,
    always_listed: bool = False,
) -> 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.

The shape (LIST vs RETRIEVE) is read from selector.kind and drives the many= flag on output_serializer at dispatch. RETRIEVE is the typical case for a URI-template lookup.

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,
    always_listed: bool = False,
) -> 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,
    include_structured_content: bool | None = None,
    include_output_schema: bool | None = None,
    argument_binding: ArgumentBinding = ArgumentBinding.DATA_ONLY,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
    spec_kwargs_provides: tuple[str, ...] = (),
) -> 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,
    kind: SelectorKind | None = None,
    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,
    include_structured_content: bool | None = None,
    include_output_schema: bool | None = None,
    argument_binding: ArgumentBinding = ArgumentBinding.MERGE,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
    spec_kwargs_provides: tuple[str, ...] = (),
) -> 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.

kind is required when spec is omitted (the decorator auto-constructs a :class:SelectorSpec and the spec's own kind field is mandatory). When spec is supplied, kind is read from spec.kind and any value passed here is ignored.

resource 

resource(
    *,
    uri_template: str,
    kind: SelectorKind | None = None,
    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,
    always_listed: bool = False,
) -> 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.

kind is required when spec is omitted; otherwise it comes from spec.kind and any value passed here is ignored.

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,
    always_listed: bool = False,
) -> 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.