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 the admin.site.urls way — .urls is a namespaced (patterns, app_name, namespace) triple path() mounts directly (no include()):

urlpatterns = [path("mcp/", server.urls)]
# reverse("mcp:endpoint") · reverse("mcp:protected-resource-metadata")

config property

config: MCPConfig

This server's resolved scalars — a frozen snapshot taken at construction.

urls property

urls: tuple[list[URLPattern], str, str]

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

Returns the namespaced (patterns, app_name, namespace) triple path() mounts directly — path("mcp/", server.urls), the admin.site.urls idiom — so the endpoints reverse within the namespace (reverse("mcp:endpoint")). 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: tuple[list[URLPattern], str, str]

Async URL patterns for ASGI deployments.

The namespaced triple (like :attr:urls), but 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,
    meta: dict[str, Any] | None = None,
    ui: UIToolMeta | None = None,
    include_structured_content: bool | None = None,
    include_output_schema: bool | None = None,
    argument_binding: ArgumentBinding = ArgumentBinding.BUNDLE,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
    spec_kwargs_provides: tuple[str, ...] = (),
    url_kwargs: tuple[UrlKwarg, ...] = (),
) -> 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.

meta is the base protocol's generic _meta bundle: an open extension namespace emitted verbatim under the "_meta" key of this tool's tools/list entry (omitted entirely when empty). It is not the annotations hint bundle — those are a closed, spec-defined set of client hints; _meta is where protocol extensions put their own keys. Passed through as given: this layer neither validates the keys nor reserves any.

ui links this tool to an interactive view registered with :meth:register_ui_resource, so a host renders the result inline instead of showing raw JSON. The view must already be registered on this server, and the tool must emit structuredContent — that is what the view renders from — or the link is refused at registration rather than shipping a view that comes up blank.

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,
    meta: dict[str, Any] | None = None,
    ui: UIToolMeta | 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.SPREAD_AUTHOR_WINS,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
    spec_kwargs_provides: tuple[str, ...] = (),
    url_kwargs: tuple[UrlKwarg, ...] = (),
) -> 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 spec.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 no spec.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.

Filtering is declared on the spec, not here: set SelectorSpec.filter_set (djangorestframework-services 0.18+) and both the HTTP and MCP transports honour it. It requires the [filter] extra (django-filter); schema generation surfaces a clear ImportError if a spec carries a filter_set without the package installed. ordering_fields / paginate stay here — they are MCP pipeline mechanics with no spec analogue.

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 (spec.filter_set / ordering_fields / paginate) and renders with many=True; RETRIEVE rejects those pipeline knobs at registration and renders the result with many=False.

meta is the generic _meta bundle for this tool's tools/list entry, and ui links it to an interactive view — both as on :meth:register_service_tool.

register_specs

register_specs(
    registry: SpecRegistry, *, overrides: Mapping[str, Mapping[str, Any]] | None = None
) -> tuple[ToolBinding | SelectorToolBinding, ...]

Register every spec in a SpecRegistry as a tool, in order.

A project exposing the same operations over more than one transport keeps its spec set in a :class:~rest_framework_services.registry.spec_registry.SpecRegistry (djangorestframework-services 0.27+) so each transport reads one source instead of enumerating the specs again. This is the MCP end of that: it walks the registry and calls :meth:register_service_tool / :meth:register_selector_tool per entry, discriminating on the spec type.

It is a source for this server's own ToolRegistry, not a replacement for it — every tool still lands as a normal binding, and names still share the one tool namespace (a collision raises, as always). The spec registry carries only what is invariant across transports (which spec, its canonical name, its tags); every MCP knob stays here, per tool, via overrides::

server.register_specs(
    registry.by_tag("public"),
    overrides={
        "list_orders": {"paginate": True, "ordering_fields": ["created_at"]},
        "refund_order": {"annotations": {"destructiveHint": True}},
    },
)

overrides maps a registered name to the keyword arguments handed to that entry's registration method. It stays a plain mapping rather than a dataclass because the two methods take different knobs — a single record would duplicate both signatures and drift from them. The keys are therefore checked against each method's own signature, which means a knob used on the wrong spec kind (paginate on a ServiceSpec) raises :exc:TypeError from that method. An overrides key naming a spec the registry doesn't hold raises :exc:ValueError here — that is a typo, not an intentional no-op.

Registration is not transactional: a failure partway leaves the earlier entries registered. That is harmless in the intended use — registration happens at configuration time, so a raise aborts startup anyway.

Returns the bindings in registration order, mirroring the per-tool methods that each return theirs.

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,
    meta: dict[str, Any] | None = None,
    ui: UIToolMeta | 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.

meta is the generic _meta bundle for this tool's tools/list entry, and ui links it to an interactive view — both as on :meth:register_service_tool.

call_tool

call_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    *,
    user: Any,
    request: Any = None,
) -> ToolResult

Invoke a registered spec-backed tool off the HTTP / JSON-RPC path.

The blessed, transport-neutral entry point: hand a tool name and a flat arguments dict (the role request.data / query params play on HTTP) plus the acting user, and get back the same :class:ToolResult the wire handlers build — without going through JSON-RPC. An in-process consumer (the django-ag-ui bridge, a Pydantic-AI toolset, a management command) calls this instead of re-implementing dispatch.

Built on the sister repo's dispatch_spec / render_spec_output / enforce_permissions, so the spec core (instance resolution, input validation, the service / selector run, the output-selector re-fetch, queryset shaping incl. filter_set, and the retrieve nullability contract) is shared with every other transport rather than reproduced.

This is the spec core only: it honours the binding's argument_binding / unknown_arguments policies and the spec's permission_classes (object-level checks included), but does not layer on the read-shaped transport extras — pagination, ordering, and a selector binding's MCP-only input_serializer — nor the transport-level MCP permissions / rate limits. For those (and for tool listing), use the full in-process transport surface, :meth:acall_tool / :meth:list_tools. Chain tools are unsupported — they orchestrate several specs and raise :class:TypeError.

Raises :class:KeyError when no tool is registered under name.

list_tools

list_tools(
    cursor: str | None = None,
    *,
    user: Any,
    request: Any = None,
    scopes: Sequence[str] | None = None,
) -> dict[str, Any] | JsonRpcError

List the tools this server exposes, exactly as the wire would.

The in-process twin of a tools/list request: returns one page of the tool catalog with the same merged inputSchema the HTTP transport advertises (serializer fields plus a selector tool's filter / ordering / pagination arguments and the additionalProperties policy), the same per-caller listing-permission filter (FILTER_LISTINGS_BY_PERMISSIONS), and the same opaque-cursor pagination — pass the returned nextCursor back to fetch the next page. A :class:JsonRpcError signals a bad cursor.

scopes are the caller's granted scopes; pass them so a ScopeRequired-gated tool is visible under FILTER_LISTINGS_BY_PERMISSIONS exactly as it would be on the wire.

Unlike :meth:call_tool (the spec core), this is the full transport surface — the entry point for an in-process consumer (the django-ag-ui bridge, a Pydantic-AI toolset) that must mirror what a remote MCP client would see. Under an event loop use :meth:alist_tools — a listing permission filter that hits the DB raises SynchronousOnlyOperation from a sync call on the loop.

alist_tools async

alist_tools(
    cursor: str | None = None,
    *,
    user: Any,
    request: Any = None,
    scopes: Sequence[str] | None = None,
) -> dict[str, Any] | JsonRpcError

Async :meth:list_tools — safe to call from an event loop.

Listing itself is pure Python, but the per-caller permission filter (FILTER_LISTINGS_BY_PERMISSIONS) may run a DB-backed check (e.g. DjangoPermRequired → user.has_perm), which raises SynchronousOnlyOperation when reached synchronously from within an event loop — the exact context an async in-process consumer runs in. The whole sync handler therefore runs in Django's thread-sensitive executor.

acall_tool async

acall_tool(
    name: str,
    arguments: dict[str, Any] | None = None,
    *,
    user: Any,
    request: Any = None,
    scopes: Sequence[str] | None = None,
) -> dict[str, Any] | JsonRpcError

Invoke a tool off the HTTP path with full transport semantics (async).

The in-process twin of a tools/call request: routes through the same async handler the wire uses, so the transport-level MCP permissions and rate limits, the selector post-fetch pipeline (filter / order / paginate), a selector binding's MCP-only input_serializer, chain tools, and the output format all apply — everything :meth:call_tool (the spec core) deliberately omits. Returns the wire's result payload (a dict carrying content / structuredContent / isError), or a :class:JsonRpcError for a protocol fault (unknown tool, malformed arguments shape, denied permission).

arguments is the flat dict that request.data / query params play on HTTP; user is the acting user and request the originating Django request when there is one (a minimal request is synthesised otherwise, mirroring :meth:call_tool). scopes are the caller's granted scopes, populating the synthetic token so a ScopeRequired-gated tool is invokable in-process just as it is on the wire.

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",
    encoding: ResourceEncoding = ResourceEncoding.JSON,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
    meta: 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.

meta is the generic _meta bundle (see :meth:register_service_tool) for this resource's listing entry — resources/list for a concrete URI, resources/templates/list for a template — and for the contents block resources/read returns.

encoding decides how the selector's value becomes the read body: JSON (the default) pretty-prints it, TEXT returns it verbatim. Anything whose mime_type is not JSON — Markdown, CSV, plain text — wants TEXT, or the document comes back wrapped in a quoted string literal. For an HTML view use :meth:register_ui_resource, which sets both.

register_ui_resource

register_ui_resource(
    *,
    name: str,
    uri: str,
    template_name: str | None = None,
    html: str | None = None,
    selector: Callable[[], str] | None = None,
    description: str | None = None,
    title: str | None = None,
    ui: UIResourceMeta | None = None,
    mime_type: str = UI_RESOURCE_MIME_TYPE,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
    meta: dict[str, Any] | None = None,
    always_listed: bool = False,
) -> ResourceBinding

Register an interactive HTML view (an MCP App) as a resource.

A tool links to the view and a host renders it inline in the chat, inside a sandboxed iframe it constructs itself. This server's whole job is to declare: serve the document, and describe what it needs in _meta. The iframe, the CSP enforcement and the ui/* postMessage bridge are the host's, and are deliberately not implemented here.

Give exactly one content source — template_name (a Django template, the idiomatic choice), html (a literal document), or selector (a zero-argument callable returning one).

Keep tenant data out of the view. Hosts may prefetch and cache a view before any tool call, so it is a shell that hydrates itself at runtime from tool results — which is also why the template renders with no context. This is a house rule rather than a spec rule, and it is the one thing a Django author's instinct gets wrong, because rendering the queryset into the template is normally the right answer.

ui= is the typed :class:UIResourceMeta — CSP origins, browser permissions, publisher domain, border preference — which serialises into _meta under the extension's key. meta= remains available for other extensions; passing both ui= and that same key inside meta= raises, rather than letting one silently win.

The result is an ordinary :class:ResourceBinding, so it shares one URI namespace with data resources (a collision raises as always), appears in resources/list, and honours permissions / always_listed. Views default to unguarded — the MCP session is already authenticated and a view is a static asset, not tenant data.

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,
    meta: 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.

meta is the generic _meta bundle for this prompt's prompts/list entry — see :meth:register_service_tool.

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,
    meta: dict[str, Any] | None = None,
    ui: UIToolMeta | None = None,
    include_structured_content: bool | None = None,
    include_output_schema: bool | None = None,
    argument_binding: ArgumentBinding = ArgumentBinding.BUNDLE,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
    spec_kwargs_provides: tuple[str, ...] = (),
    url_kwargs: tuple[UrlKwarg, ...] = (),
) -> 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,
    meta: dict[str, Any] | None = None,
    ui: UIToolMeta | 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.SPREAD_AUTHOR_WINS,
    unknown_arguments: UnknownArguments = UnknownArguments.REJECT,
    always_listed: bool = False,
    spec_kwargs_provides: tuple[str, ...] = (),
    url_kwargs: tuple[UrlKwarg, ...] = (),
) -> 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",
    encoding: ResourceEncoding = ResourceEncoding.JSON,
    permissions: list[Any] | None = None,
    rate_limits: list[Any] | None = None,
    annotations: dict[str, Any] | None = None,
    meta: 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,
    meta: 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.