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")

What every spec on the mount may read -- a tenant, a locale, a clock -- is registered once as pool_seeds=, a drf-services PoolSeeds:

server = MCPServer(
    pool_seeds=DEFAULT_POOL_SEEDS.extend(tenant=lambda *, user: user.tenant),
)

Every spec the server runs receives the seeds (tools over either transport, call_tool / acall_tool, task workers, chain steps and resource reads), every operation condition it asks is asked with them, tools/list availability included, and a registered name is reserved: a client argument cannot occupy it, and a UrlKwarg, QueryParam or URI-template variable naming it is refused at registration.

What the server tells a model in its own words -- the handle description and line, a paged tool's scope sentence, the missing-argument message -- is conventions=, an AgentConventions. Change one sentence and keep the rest:

server = MCPServer(
    conventions=AgentConventions(handle_line="Pass ids to other tools; never show them."),
)

The server still decides where each sentence appears; the conventions decide what it says, on every route the seeds above reach.

config `property` ¶

config: MCPConfig

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

pool_seeds `property` ¶

pool_seeds: PoolSeeds

The pool_seeds= every dispatch and condition on this server receives.

conventions `property` ¶

conventions: AgentConventions

The conventions= every listing and refusal on this server is worded with.

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 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 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 asgiref.sync.sync_to_async, so a fully sync stack still works; async-native ones 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,
    icons: tuple[Icon, ...] = (),
    display_name: str | None = None,
    display_description: str | None = None,
    output_format: OutputFormat | str = OutputFormat.JSON,
    content_kind: ToolContentKind = ToolContentKind.TEXT,
    invalidates: tuple[str, ...] | list[str] = (),
    task_policy: TaskPolicy = TaskPolicy.FORBIDDEN,
    content_mime_type: str | 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,
    agent_contract: OfflineContract | 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, ...] = (),
    query_params: tuple[QueryParam, ...] = (),
    max_result_bytes: int | None | UnsetType = UNSET,
    dispatch_timeout: float | None | UnsetType = UNSET,
    server_arguments: ServerArguments | None = None,
) -> ToolBinding

Register a ServiceSpec as an MCP mutation tool.

The dispatch pipeline runs input_serializer → run_service(atomic) → output_selector? → output_serializer, so this is the surface for side-effecting operations. For read-shaped ones (list/retrieve with optional filtering / ordering / pagination) use register_selector_tool instead.

meta is the base protocol's generic _meta bundle, emitted verbatim under the "_meta" key of this tool's tools/list entry and omitted when empty. It is not the annotations hint bundle — those are a closed, spec-defined set of client hints, while _meta is where protocol extensions put their own keys. Passed through as given: no key is validated or reserved here.

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

The route captures and query values a caller with no HTTP request has to send come from the spec's own reads, and the output fields an agent sees from its field_audiences: the declaration is made once, on the spec, and this registrar, the decorator and register_specs all read it, so one spec is the same tool however it is mounted.

server_arguments names declared parameters the server answers for, so no caller does: a ServerArguments from django_service_specs, whose values are fixed for the life of the binding and whose withheld names leave the callable to its own default. Both are dropped from the advertised schema and from required; a caller's value for one is dropped before dispatch and logged by name. It is checked at registration, so an undeclared name, a value the declared type refuses or a withheld name with no default raises ImproperlyConfigured naming the tool. A value computed per request is not something this argument carries.

url_kwargs, query_params and agent_contract are deprecated: each declares beside the spec what the spec declares itself, and each warns, naming its replacement. An explicit url_kwargs / query_params replaces the spec's reads of that kind in what the tool advertises and routes; agent_contract replaces all of them and the field audiences. Neither takes a channel away from the dispatch, which still reads the spec's declaration; only server_arguments' withheld does.

register_selector_tool ¶

register_selector_tool(
    *,
    name: str,
    spec: SelectorSpec,
    description: str | None = None,
    title: str | None = None,
    icons: tuple[Icon, ...] = (),
    display_name: str | None = None,
    display_description: str | None = None,
    input_serializer: type | None = None,
    output_format: OutputFormat | str = OutputFormat.JSON,
    content_kind: ToolContentKind = ToolContentKind.TEXT,
    task_policy: TaskPolicy = TaskPolicy.FORBIDDEN,
    content_mime_type: str | 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,
    agent_contract: OfflineContract | None = None,
    ui: UIToolMeta | 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, ...] = (),
    query_params: tuple[QueryParam, ...] = (),
    max_result_bytes: int | None | UnsetType = UNSET,
    dispatch_timeout: float | None | UnsetType = UNSET,
    max_page_size: int | None | UnsetType = UNSET,
    server_arguments: ServerArguments | None = None,
) -> SelectorToolBinding

Register a SelectorSpec as an MCP read tool.

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

arguments → validate(merged inputSchema)
          → run_selector
          → FilterSet(data=...).qs    (if spec.filter_set set; it
                                       orders too when it declares an
                                       OrderingFilter)
          → paginate                  (if paginate=True)
          → output_serializer(many=True)
          → ToolResult

Each knob is optional; with none of them set the tool is a plain RPC read against the selector.

Filtering and ordering are declared on the spec, not here: set SelectorSpec.filter_set and both the HTTP and MCP transports honour it, ordering included via an OrderingFilter. It requires the [filter] extra (django-filter), and schema generation raises a clear ImportError without it. paginate stays here, being an MCP pipeline mechanic with no spec analogue.

The shape comes from spec.kind: LIST runs the full post-fetch pipeline and renders with many=True; RETRIEVE rejects paginate at registration and renders 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 register_service_tool.

The route captures and query values a caller with no HTTP request has to send come from the spec's own reads, and the output fields an agent sees from its field_audiences: the declaration is made once, on the spec, and this registrar, the decorator and register_specs all read it, so one spec is the same tool however it is mounted.

server_arguments names declared parameters the server answers for, so no caller does: a ServerArguments from django_service_specs, whose values are fixed for the life of the binding and whose withheld names leave the callable to its own default. Both are dropped from the advertised schema and from required; a caller's value for one is dropped before dispatch and logged by name. It is checked at registration, so an undeclared name, a value the declared type refuses or a withheld name with no default raises ImproperlyConfigured naming the tool. A value computed per request is not something this argument carries.

url_kwargs, query_params and agent_contract are deprecated: each declares beside the spec what the spec declares itself, and each warns, naming its replacement. An explicit url_kwargs / query_params replaces the spec's reads of that kind in what the tool advertises and routes; agent_contract replaces all of them and the field audiences. Neither takes a channel away from the dispatch, which still reads the spec's declaration; only server_arguments' withheld does.

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 SpecRegistry so each transport reads one source. This walks it and calls register_service_tool / register_selector_tool per entry, discriminating on the spec type.

It is a source for this server's own ToolRegistry, not a replacement — every tool lands as a normal binding sharing the one tool namespace (a collision raises, as always). The registry carries only what is invariant across transports; every MCP knob stays here, per tool, via overrides:

server.register_specs(
    registry.by_tag("public"),
    overrides={
        "list_orders": {"paginate": True, "max_page_size": 25},
        "refund_order": {"annotations": {"destructiveHint": True}},
    },
)

Each spec's own reads and field_audiences are read by the registrar, as on a direct registration. server_arguments is an override like any other, and the one to reach for when this mount answers an argument itself:

server.register_specs(
    registry,
    overrides={
        "list_documents": {
            "server_arguments": ServerArguments(values={"project_pk": 7}),
        },
    },
)

A url_kwargs, query_params or agent_contract override is deprecated and warns, as the registrar arguments do.

Keys are checked against the target method's own signature, so a knob used on the wrong spec kind (paginate on a ServiceSpec) raises TypeError from there. An overrides key naming a spec the registry doesn't hold raises ValueError here — that is a typo, not an intentional no-op.

Registration is not transactional: a failure partway leaves the earlier entries registered, which is harmless at configuration time because a raise aborts startup anyway.

Returns the bindings in registration order.

register_chain_tool ¶

register_chain_tool(
    *,
    name: str,
    steps: list[ChainStep] | tuple[ChainStep, ...],
    description: str | None = None,
    title: str | None = None,
    icons: tuple[Icon, ...] = (),
    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,
    content_kind: ToolContentKind = ToolContentKind.TEXT,
    invalidates: tuple[str, ...] | list[str] = (),
    task_policy: TaskPolicy = TaskPolicy.FORBIDDEN,
    content_mime_type: str | 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,
    agent_contract: OfflineContract | 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,
    max_result_bytes: int | None | UnsetType = UNSET,
    dispatch_timeout: float | None | UnsetType = UNSET,
) -> ChainToolBinding

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

Each 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.

atomic=True 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 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 judged up front, each step's against a stand-in carrying its alias, so a step permission refusing on the chain's arguments blocks the whole chain before any step runs. Each step judges them again against its own view before its inputs run and its target is looked up, and only the object-level half on the target it resolves.

Chains deliberately do not run the selector post-fetch pipeline (filter / order / paginate); for that, expose the selector as its own 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 register_service_tool.

agent_contract is the same carrier the two spec registrars take, and a chain's only route to it: a chain has no registry entry to inherit from. Only its field_audiences apply -- a chain declares its arguments through its steps.

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 transport-neutral entry point: hand a tool name, a flat arguments dict (the role request.data / query params play on HTTP) and the acting user, and get back the same ToolResult the wire handlers build. An in-process consumer calls this instead of re-implementing dispatch.

This is the spec core only — instance resolution, input validation, the service / selector run, the output-selector re-fetch, queryset shaping including filter_set, and the retrieve nullability contract, shared with every other transport rather than reproduced. It honours the binding's argument_binding / unknown_arguments policies and the spec's permission_classes (object-level checks included), but not the read-shaped transport extras — pagination, a selector binding's MCP-only input_serializer — nor the transport-level MCP permissions and rate limits. A FilterSet's ordering is not one of those extras: it is applied here with the rest of the filtering. For those, and for tool listing, use acall_tool / list_tools. Chain tools orchestrate several specs and raise TypeError here.

Raises 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,
    include_unavailable: bool = False,
) -> dict[str, Any] | JsonRpcError

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

The in-process twin of a tools/list request: 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 for the next page. A 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. A tool whose operation-scope affordances condition is unmet for user right now is left out, as on the wire, unless include_unavailable=True, which lists every tool the caller may see and asks no condition. That is for a consumer that builds its tool definitions once and asks unavailable_tools each step: it needs a definition for a tool that is unavailable when it lists and becomes available later. The wire has no such parameter.

Unlike call_tool (the spec core) this is the full transport surface. Under an event loop use alist_tools — a listing permission filter or an affordance condition 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,
    include_unavailable: bool = False,
) -> dict[str, Any] | JsonRpcError

Async list_tools — safe to call from an event loop.

Listing itself is pure Python, but the per-caller permission filter (FILTER_LISTINGS_BY_PERMISSIONS) and a tool's operation-scope affordances condition may each run a DB-backed check, which raises SynchronousOnlyOperation when reached synchronously from within an event loop. The whole sync handler therefore runs in Django's thread-sensitive executor.

unavailable_tools ¶

unavailable_tools(
    *, user: Any, request: Any = None, scopes: Sequence[str] | None = None
) -> dict[str, Affordance]

The tools list_tools leaves out for this caller right now, and why.

Maps each omitted tool's name to the operation-scope Affordance whose condition is unmet: its code and its reason, the sentence written for people and models alike. For a chain it is the first unmet one in step order. The answer comes from the same pass tools/list makes, over the same tools the caller may see (FILTER_LISTINGS_BY_PERMISSIONS included), so a name is here exactly when the listing omits it, and a tool hidden from this caller is never named with its reason.

For an in-process consumer that lists once with include_unavailable=True and asks this each step: it offers its model what a fresh tools/list would, and can tell the model why the rest are missing, which a listing cannot, since it only leaves them out. always_listed=True tools are never here, as they are never left out. Advisory like the listing: acall_tool still enforces every condition.

Under an event loop use aunavailable_tools; a condition or listing permission that queries raises SynchronousOnlyOperation from here.

aunavailable_tools `async` ¶

aunavailable_tools(
    *, user: Any, request: Any = None, scopes: Sequence[str] | None = None
) -> dict[str, Affordance]

Async unavailable_tools -- safe to call from an event loop.

Runs in Django's thread-sensitive executor, as alist_tools does, since a condition or a listing permission may query. A server none of whose tools declares an operation-scope condition answers {} without the hop: that is decided from the declarations alone, and a consumer asking on every model step should not pay a thread switch per step to learn it.

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 call_tool omits. Returns the wire's result payload (a dict carrying content / structuredContent / isError), or a JsonRpcError for a protocol fault (unknown tool, malformed arguments shape, denied permission).

request is the originating Django request when there is one; a minimal one is synthesised otherwise, mirroring call_tool. scopes populate the synthetic token so a ScopeRequired-gated tool is invokable in-process just as it is on the wire.

run_task ¶

run_task(task_id: str) -> None

Execute a queued task. This is what a worker calls.

The other end of task_executor.enqueue, and the whole public surface of the worker side:

@shared_task
def run_mcp_task(task_id: str) -> None:
    my_server.run_task(task_id)

Everything it needs comes out of the store: the tool, the arguments, and the authorization context to re-check them under. Nothing is returned — the client learns the outcome by polling tasks/get.

Safe to call for an id that is unknown, already claimed or already finished: each is a no-op. Queues deliver at least once, and a retried delivery must not run a mutation twice.

Raises when the server has no task store: a worker calling this on a server that cannot run tasks would otherwise fail as silence — the job "succeeds" and the client polls a handle forever.

notify_resource_updated `async` ¶

notify_resource_updated(uri: str) -> int

Tell every subscriber watching uri that it changed.

The explicit trigger, and the one that always works. Call it from wherever the write actually happens — a management command, a Celery job, a save() override, a signal handler you wrote yourself:

await server.notify_resource_updated(f"invoices://{invoice.pk}")

Returns how many subscribers were reached — a diagnostic, not a guarantee: 0 means nobody was listening, which is the ordinary case and not an error. Notifications are best-effort by design; a client that misses one re-reads the resource.

A URI, not a template. Publish the concrete URI that changed, and the collection URI too if watchers of the collection should hear about it — matching is exact, deliberately (see topic_for_resource).

Publish after the transaction commits. Inside transaction.atomic() this announces a change that may still roll back, and a subscriber that re-reads immediately sees the old value — worse than no notification at all. transaction.on_commit is the Django-native answer.

notify_list_changed `async` ¶

notify_list_changed(kind: NotificationKind) -> int

Tell subscribers that one of the catalogs changed.

Registration happens once at configuration time, so the set of registered bindings is fixed for the life of the process. Two things still change a catalog under a running client: a server that registers tools from data — a plugin loader, a per-tenant catalog — and a tool whose operation-scope affordances condition flips, which tools/list answers on every request and so starts or stops listing it. Call this with TOOLS_LIST_CHANGED after the event that flipped the condition, once it has committed. It reaches only clients this server can push to: a subscription broker configured and a modern-era caller listening, the same two things listChanged is advertised on.

register_resource ¶

register_resource(
    *,
    name: str,
    uri_template: str,
    selector: SelectorSpec,
    description: str | None = None,
    title: str | None = None,
    icons: tuple[Icon, ...] = (),
    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,
    cache_ttl_ms: int | UnsetType = UNSET,
    completions: dict[str, Callable[..., Any]] | None = None,
) -> ResourceBinding

Register a SelectorSpec as an MCP resource.

selector.selector is the callable dispatched at resources/read time; selector.output_serializer fills in when the explicit output_serializer= kwarg is absent (that kwarg wins); selector.kwargs becomes the binding's per-request kwargs provider.

A bare callable is not accepted here — wrap it in SelectorSpec(selector=fn), or use the decorator form resource, which wraps it automatically.

The shape comes 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 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 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 register_ui_resource, which sets both.

A uri_template variable is a caller-controlled name that reaches the selector's kwarg pool, so one named after a dispatcher seed (user, request, data …) or declared twice raises here rather than letting a URI segment stand in for the authenticated identity.

A resource with no permissions at all is refused for the same reason a tool is — see REQUIRE_TOOL_PERMISSIONS. The same selector exposed as a resource is as reachable as it is exposed as a tool.

register_ui_resource ¶

register_ui_resource(
    *,
    name: str,
    uri: str,
    template_name: str | None = None,
    body_template_name: str | None = None,
    diagnostics: bool | None = None,
    html: str | None = None,
    selector: Callable[[], str] | None = None,
    description: str | None = None,
    title: str | None = None,
    icons: tuple[Icon, ...] = (),
    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,
    cache_ttl_ms: int | UnsetType = UNSET,
) -> 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. The iframe and the CSP enforcement are the host's and are deliberately not implemented here.

The ui/* postMessage bridge is not in that list, and reading it as though it were is what this parameter set exists to prevent. The bridge has two ends: the host runs its end, and the document runs the other. That second end is a view's mandatory startup handshake, every one of its failure modes is silent, and one of them leaves the frame hidden with the view unable to say why.

So give exactly one content source, and prefer the first:

body_template_name — a Django template holding the view's markup only. The package wraps it in a document whose bridge is already written and already correct. diagnostics= decides whether that bridge writes a protocol failure into the document as well as logging it; None follows settings.DEBUG, because the text is written for whoever wrote the view and a rendered view's audience is whoever is using the product.
template_name — a Django template holding a whole document, whose bridge is then yours to write.
html — a literal document, same.
selector — a callable returning one, which must take no arguments: a selector that declares a parameter is refused at registration, because the read path would fill it from a pool carrying request and user.

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.

ui= is the typed 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 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,
    icons: tuple[Icon, ...] = (),
    arguments: list[PromptArgument] | None = None,
    completions: dict[str, Callable[..., Any]] | 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 PromptMessage, or a coroutine yielding any of those — the dispatch layer normalises the result.

request and user are seeded over the client's arguments at prompts/get, so an argument named after one of them never reaches render in the seed's place.

A prompt with no permissions at all is refused like a tool — see REQUIRE_TOOL_PERMISSIONS. A render callable reads whatever its author gave it access to, so nothing about a prompt makes it safe by construction.

meta is the generic _meta bundle for this prompt's prompts/list entry — see 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,
    icons: tuple[Icon, ...] = (),
    output_format: OutputFormat | str = OutputFormat.JSON,
    content_kind: ToolContentKind = ToolContentKind.TEXT,
    invalidates: tuple[str, ...] | list[str] = (),
    task_policy: TaskPolicy = TaskPolicy.FORBIDDEN,
    content_mime_type: str | 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,
    agent_contract: OfflineContract | 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, ...] = (),
    query_params: tuple[QueryParam, ...] = (),
    server_arguments: ServerArguments | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator form of register_service_tool.

If spec is supplied it is used verbatim; otherwise a ServiceSpec is constructed from the keyword arguments. The original function is returned unchanged, so it stays callable from Python without going through the MCP transport. Every keyword means what it does on register_service_tool, server_arguments included, and is checked by the same registration.

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,
    icons: tuple[Icon, ...] = (),
    output_format: OutputFormat | str = OutputFormat.JSON,
    content_kind: ToolContentKind = ToolContentKind.TEXT,
    task_policy: TaskPolicy = TaskPolicy.FORBIDDEN,
    content_mime_type: str | 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,
    agent_contract: OfflineContract | None = None,
    ui: UIToolMeta | 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, ...] = (),
    query_params: tuple[QueryParam, ...] = (),
    server_arguments: ServerArguments | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator form of register_selector_tool.

If spec is supplied it is used verbatim; otherwise a SelectorSpec is constructed from the wrapped function and the keyword arguments. The original function is returned unchanged, so it stays 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. Every other keyword means what it does on register_selector_tool, server_arguments included, and is checked by the same registration.

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,
    icons: tuple[Icon, ...] = (),
    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,
    cache_ttl_ms: int | UnsetType = UNSET,
    completions: dict[str, Callable[..., 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 SelectorSpec is constructed from the wrapped function and the keyword arguments. The original function is returned unchanged, so it stays 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,
    icons: tuple[Icon, ...] = (),
    arguments: list[PromptArgument] | None = None,
    completions: dict[str, Callable[..., Any]] | 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 connected — a missed push is not generally an error, since clients pull state via tools/call round-trips. The broker enforces one subscriber per session: re-subscribing replaces the old queue silently.

With a SSEReplayBuffer configured the payload is recorded before publishing, so the frame carries an event ID the SSE generator emits on the wire (id: <id>\ndata: <payload>\n\n) and a later reconnect with Last-Event-ID drains what it missed before resuming live mode. Without a buffer there are no id: lines and resume is disabled.

Multi-process deployments need an out-of-process broker to fan out across workers; the in-process broker only sees its own.

AgentConventions `dataclass` ¶

The sentences this server writes for a model, one field per sentence.

Passed as MCPServer(conventions=...), per server, because two servers in one project can be talking to different readers. AgentConventions() is the default, and its fields carry the wording a server uses when it is given none; change one field and every other sentence stays the package's, so a later correction to a line you did not override still reaches you.

The server still decides whether each sentence appears: a tool with no handle gets no handle line, a tool that returns no page no scope sentence. A field changes what it says, never when. None drops a sentence where dropping one leaves a well-formed answer.

Every field is a str.format template, rendered wherever it lands whether or not it has a placeholder, so a literal brace is written twice, {{ or }}, in every field alike. The placeholders a field accepts are the ones listed on it, and only missing_arguments accepts one. Anything the server could not render raises ImproperlyConfigured naming the field when the conventions are built, so a typo fails when the server is, rather than inside the first call that reaches it. This is the Pydantic-AI toolset's rule for its AgentConventions, so one template reads the same on both transports.

Some of these sentences state facts about how the server behaves: that a selection applies to each item and never to the envelope, that the envelope's keys are items, page, totalPages and hasNext. Wording you supply owns keeping those facts true.

Raises:

Type	Description
`ImproperlyConfigured`	A field uses a placeholder it does not accept, is not a valid format string, cannot be rendered with the values it will be given (`{names:q}`), is neither a string nor `None`, or is `missing_arguments` set to `None` or to an empty or whitespace-only string.

handle_field_description `class-attribute` `instance-attribute` ¶

handle_field_description: str | None = HANDLE_DESCRIPTION

The outputSchema description of a handle field that declares none of its own, beside the field in every tool's output schema. None leaves such a field undescribed. A FieldMarking.handle("...") keeps its own words whatever this says. No placeholders.

handle_line `class-attribute` `instance-attribute` ¶

handle_line: str | None = HANDLE_LINE

The line appended to the description of a tool whose output carries a handle, after Identify records by `<label>`. when the output serializer declares a label. None drops the line and that prefix with it. No placeholders.

query_param_on_pages `class-attribute` `instance-attribute` ¶

query_param_on_pages: str | None = PAGED_QUERY_PARAM_SCOPE

What a read-shaping QueryParam applies to on a tool that returns a page. Appended to the param's description in the tool's inputSchema, and to the isError message when a value it supplied is refused while that tool's result is rendered. None drops it from both. No placeholders.

missing_arguments `class-attribute` `instance-attribute` ¶

missing_arguments: str = MISSING_ARGUMENTS

The message of a validation_error result for a call that left out arguments the tool's selectors cannot run without: a selector parameter with no default, a service tool's target lookup, a UrlKwarg(required=True). {names} is the missing names, sorted, each in backticks, joined with ", "; the detail stays keyed by each name. An input serializer's own refusal, a missing field included, keeps "Invalid arguments". Never None, empty or only whitespace: it is the result's whole message.

MCPServer¶

MCPServer ¶

config property ¶

pool_seeds property ¶

conventions property ¶

urls property ¶

async_urls property ¶