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 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,
) -> 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.

agent_contract carries what a caller with no HTTP request has to be told -- the URL kwargs, query params and field-audience overrides the URLconf and query string give an HTTP caller for free. register_specs passes each entry's own, so the declaration is made once and every agent transport reads it; an explicit url_kwargs / query_params here wins over it.

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,
) -> 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.

agent_contract carries what a caller with no HTTP request has to be told -- the URL kwargs, query params and field-audience overrides the URLconf and query string give an HTTP caller for free. register_specs passes each entry's own, so the declaration is made once and every agent transport reads it; an explicit url_kwargs / query_params here wins over it.

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 entry's OfflineContract comes across as the mount's default — the url_kwargs, query_params and field_audiences an off-HTTP caller needs and an HTTP one gets from the URLconf and query string for free. A per-tool url_kwargs / query_params override wins over it; overriding agent_contract itself replaces it outright, which is the only way to register an entry with fewer channels than it declares.

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 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 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,
) -> 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.

Unlike call_tool (the spec core) this is the full transport surface. Under an event loop use 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 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, which raises SynchronousOnlyOperation when reached synchronously from within an event loop. 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 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.

Rarely needed: registration happens once at configuration time, so a catalog is fixed for the life of the process. It exists for the server that registers tools from data — a plugin loader, a per-tenant catalog — where the list genuinely can change under a running client.

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, ...] = (),
) -> 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.

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, ...] = (),
) -> 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.

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.