def create_scoped_mcp_app( # noqa: PLR0913 -- explicit host-owned binding contract
*,
allowed_tools: frozenset[str],
manifest: DeclarativeToolManifest | None,
invoke: Callable[[str, dict[str, object]], Awaitable[ToolResult]],
authorize: Callable[[Scope], bool | Awaitable[bool]],
policy_revision: str,
tools: Mapping[str, AXMTool] | None = None,
write_contract: WriteContract | None = None,
) -> ASGIApp:
"""Build a host-owned scope, without modifying the shared MCP server.
Mount at a capability-specific prefix: /mcp and adjacent /health. The
authorization callback receives the actual ASGI scope and runs before every
HTTP request and tool dispatch. Client headers never supply or enlarge tools
or the write contract. A stateless request owns and closes its SDK manager;
dynamic mounts therefore need no per-capability startup coordination.
"""
if not policy_revision or not isinstance(allowed_tools, frozenset):
raise ValueError("immutable positive tool list and policy revision required")
if allowed_tools & _FACADE:
raise ValueError("declare target names, not facade wrappers")
source = (
discover_tools() if tools is None else cast(dict[str, ToolEntry], dict(tools))
)
catalog = _resolved_catalog(allowed_tools, manifest, source, invoke, write_contract)
handlers = _ScopedHandlers(catalog, authorize)
server: Server[object] = Server(
"AXM scoped MCP",
on_list_tools=handlers.list_tools,
on_call_tool=handlers.call_tool,
)
return _ScopedApp(
server,
authorize,
{
"kind": "axm-scoped-mcp",
"schema_version": 1,
"policy_revision": policy_revision,
"tool_scope_sha256": tool_scope_sha256(allowed_tools),
"write_contracts_enforced": write_contract is not None,
},
)