modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add a pluggable server extension API with MCP Apps (#3003)

Browse source 
Co-authored-by: Max Isbey <224885523+maxisbey@users.noreply.github.com>
This commit is contained in:
Marcelo Trylesinski 2026-06-29 11:58:05 +02:00 • committed by GitHub
commit 4b519782f1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 2598 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

485
tests/server/mcpserver/test_extension.py Normal file
View file

@ -0,0 +1,485 @@
"""Tests for the core SEP-2133 extension API (`Extension`, `MCPServer` wiring).
These exercise the closed set of extension contribution kinds - tools,
resources, request methods, and the single `tools/call` interceptor - through
the highest-level public surface (in-memory `Client`), plus the
`compose_tool_call_interceptor` helper directly.
"""
from typing import Any, Literal, cast
import mcp_types as types
import pytest
from inline_snapshot import snapshot
from mcp_types import (
METHOD_NOT_FOUND,
MISSING_REQUIRED_CLIENT_CAPABILITY,
CallToolResult,
TextContent,
)
from mcp.client.client import Client
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import (
Extension,
MethodBinding,
ResourceBinding,
ToolBinding,
compose_tool_call_interceptor,
validate_extension_identifier,
)
from mcp.server.mcpserver import Context, MCPServer, require_client_extension
from mcp.server.mcpserver.resources import TextResource
from mcp.shared.exceptions import MCPError
pytestmark = pytest.mark.anyio
_TOOL_META: dict[str, Any] = {"com.example/marker": {"v": 1}}
class _AdditiveExt(Extension):
"""Override `tools()`/`resources()` only - a purely additive extension."""
identifier = "com.example/additive"
def tools(self):
def ping() -> str:
"""Reply with pong."""
return "pong"
return [ToolBinding(fn=ping, meta=_TOOL_META)]
def resources(self):
return [ResourceBinding(resource=TextResource(uri="ext://greeting", name="greeting", text="hello"))]
class _SettingsExt(Extension):
"""Override `settings()` so the extension advertises a non-empty settings map."""
identifier = "com.example/settings"
def settings(self) -> dict[str, Any]:
return {"feature": {"enabled": True}}
class _PingParams(types.RequestParams):
pass
class _PingResult(types.Result):
pong: bool
class _PingRequest(types.Request[_PingParams, Literal["com.example/ping"]]):
method: Literal["com.example/ping"] = "com.example/ping"
params: _PingParams
async def _pong_handler(ctx: ServerRequestContext[Any, Any], params: _PingParams) -> _PingResult:
"""The shared `com.example/ping` handler (dispatched by the reachability test)."""
return _PingResult(pong=True)
class _MethodExt(Extension):
"""Override `methods()` to serve a new vendor request verb."""
identifier = "com.example/method"
def methods(self) -> list[MethodBinding]:
return [MethodBinding("com.example/ping", _PingParams, _pong_handler)]
class _ReplacingExt(Extension):
"""Override `intercept_tool_call()` to short-circuit with a fixed result."""
identifier = "com.example/replacing"
async def intercept_tool_call(
self, params: types.CallToolRequestParams, ctx: ServerRequestContext[Any, Any], call_next: CallNext
) -> HandlerResult:
return CallToolResult(content=[TextContent(type="text", text="intercepted")])
class _PassThroughExt(Extension):
"""Override `intercept_tool_call()` but always delegate to `call_next` unchanged."""
identifier = "com.example/passthrough"
async def intercept_tool_call(
self, params: types.CallToolRequestParams, ctx: ServerRequestContext[Any, Any], call_next: CallNext
) -> HandlerResult:
return await call_next(ctx)
class _DefaultExt(Extension):
"""Override nothing - relies on the base `intercept_tool_call` default (pass through)."""
identifier = "com.example/default"
class _RecordingExt(Extension):
"""Override `intercept_tool_call()` to record `(identifier, tool_name)` then pass through."""
def __init__(self, identifier: str, log: list[tuple[str, str]]) -> None:
self.identifier = identifier
self._log = log
async def intercept_tool_call(
self, params: types.CallToolRequestParams, ctx: ServerRequestContext[Any, Any], call_next: CallNext
) -> HandlerResult:
self._log.append((self.identifier, params.name))
return await call_next(ctx)
def _echo(value: str) -> str:
"""Echo the input value (shared tool body across interceptor tests)."""
return value
async def test_additive_extension_registers_its_tool_and_resource() -> None:
"""SDK-defined: an `Extension` overriding `tools()`/`resources()` surfaces both
through `MCPServer`'s normal `list_tools`/`list_resources`, and the tool's
`_meta` round-trips equal to the exact dict the binding carried (identity can't
hold - the value is JSON-serialized over the transport)."""
server = MCPServer("test", extensions=[_AdditiveExt()])
async with Client(server) as client:
tools = await client.list_tools()
resources = await client.list_resources()
called = await client.call_tool("ping", {})
assert [t.name for t in tools.tools] == ["ping"]
assert tools.tools[0].meta == _TOOL_META
assert called == snapshot(CallToolResult(content=[TextContent(text="pong")], structured_content={"result": "pong"}))
assert resources == snapshot(
types.ListResourcesResult(
resources=[types.Resource(name="greeting", uri="ext://greeting", mime_type="text/plain")]
)
)
async def test_extension_settings_advertised_under_server_capabilities() -> None:
"""SDK-defined: `settings()` rides `server/discover` and lands under
`server_capabilities.extensions[identifier]` on the modern (`auto`) path."""
server = MCPServer("test", extensions=[_SettingsExt()])
async with Client(server, mode="auto") as client:
extensions = client.server_capabilities.extensions
assert extensions == snapshot({"com.example/settings": {"feature": {"enabled": True}}})
async def test_extension_settings_dropped_on_legacy_handshake() -> None:
"""Pinned gap: the 2025 `ServerCapabilities` wire schema has no `extensions`
field, so a legacy `initialize` handshake drops the advertised extension even
though the modern `auto` path carries it."""
server = MCPServer("test", extensions=[_SettingsExt()])
async with Client(server, mode="legacy") as client:
assert client.server_capabilities.extensions is None
def test_duplicate_extension_identifier_raises() -> None:
"""SDK-defined: registering two extensions with the same `identifier` is a
construction error."""
with pytest.raises(ValueError):
MCPServer("test", extensions=[_SettingsExt(), _SettingsExt()])
async def test_extension_method_reachable_via_session_send_request() -> None:
"""SDK-defined: an `Extension` overriding `methods()` wires a new request verb
onto the low-level server, reachable through `client.session.send_request`."""
server = MCPServer("test", extensions=[_MethodExt()])
async with Client(server) as client:
request = _PingRequest(params=_PingParams())
result = await client.session.send_request(cast("types.ClientRequest", request), _PingResult)
assert result == snapshot(_PingResult(pong=True))
async def test_pass_through_interceptor_leaves_tool_result_unchanged() -> None:
"""SDK-defined: an extension whose `intercept_tool_call` delegates to
`call_next` does not alter the underlying tool's `CallToolResult`."""
server = MCPServer("test", extensions=[_PassThroughExt()])
server.tool(name="echo")(_echo)
async with Client(server) as client:
result = await client.call_tool("echo", {"value": "hi"})
assert result == snapshot(CallToolResult(content=[TextContent(text="hi")], structured_content={"result": "hi"}))
async def test_short_circuiting_interceptor_replaces_tool_result() -> None:
"""SDK-defined: an extension that returns from `intercept_tool_call` without
calling `call_next` replaces the tool's result wholesale (the tool never runs)."""
server = MCPServer("test", extensions=[_ReplacingExt()])
server.tool(name="echo", structured_output=False)(_echo)
async with Client(server) as client:
result = await client.call_tool("echo", {"value": "hi"})
assert result == snapshot(CallToolResult(content=[TextContent(text="intercepted")]))
def test_plain_extension_installs_no_tool_call_interceptor() -> None:
"""SDK-defined: an extension that does not override `intercept_tool_call` adds no
middleware - the composed interceptor exists only when at least one extension
overrides it."""
baseline = len(MCPServer("test")._lowlevel_server.middleware)
server = MCPServer("test", extensions=[_AdditiveExt()])
assert len(server._lowlevel_server.middleware) == baseline
def test_overriding_extension_installs_one_tool_call_interceptor() -> None:
"""SDK-defined: an extension that overrides `intercept_tool_call` composes exactly
one additional `tools/call` middleware."""
baseline = len(MCPServer("test")._lowlevel_server.middleware)
server = MCPServer("test", extensions=[_ReplacingExt()])
assert len(server._lowlevel_server.middleware) == baseline + 1
async def test_default_interceptor_passes_through_alongside_an_overriding_one() -> None:
"""SDK-defined: an extension that does not override `intercept_tool_call` runs the
base-class default (pass through) when another extension forces the composed
middleware to exist, leaving the tool result untouched."""
server = MCPServer("test", extensions=[_DefaultExt(), _PassThroughExt()])
server.tool(name="echo")(_echo)
async with Client(server) as client:
result = await client.call_tool("echo", {"value": "hi"})
assert result == snapshot(CallToolResult(content=[TextContent(text="hi")], structured_content={"result": "hi"}))
async def test_interceptors_run_in_registration_order_with_threaded_params() -> None:
"""SDK-defined: `compose_tool_call_interceptor` nests extensions first-outermost, so
two passing-through interceptors record in registration order, each seeing the
validated `tools/call` params (the real tool name)."""
log: list[tuple[str, str]] = []
server = MCPServer(
"test",
extensions=[_RecordingExt("com.example/first", log), _RecordingExt("com.example/second", log)],
)
server.tool(name="echo")(_echo)
async with Client(server) as client:
await client.call_tool("echo", {"value": "hi"})
assert log == [("com.example/first", "echo"), ("com.example/second", "echo")]
async def test_compose_tool_call_interceptor_passes_through_non_tools_call() -> None:
"""SDK-defined: the composed middleware is a no-op for any method other than
`tools/call` - it forwards to `call_next` without touching the interceptors."""
sentinel = types.EmptyResult()
async def call_next(ctx: ServerRequestContext[Any, Any]) -> HandlerResult:
return sentinel
middleware = compose_tool_call_interceptor([_ReplacingExt()])
ctx = ServerRequestContext(
session=cast("Any", None),
lifespan_context={},
protocol_version="2026-07-28",
method="tasks/get",
params={"taskId": "t-1"},
)
result = await middleware(ctx, call_next)
assert result is sentinel
def test_extension_subclass_without_prefixed_identifier_is_rejected_at_definition() -> None:
"""SDK-defined: SEP-2133 requires a `vendor-prefix/name` identifier, enforced when the
subclass is defined (a bare name with no prefix is a TypeError)."""
with pytest.raises(TypeError):
type("_BadExt", (Extension,), {"identifier": "noprefix"})
def test_extension_without_identifier_is_rejected_at_registration() -> None:
"""SDK-defined: a subclass that never sets `identifier` (neither class-level nor in
`__init__`) is rejected when the server applies it."""
class _NoIdExt(Extension):
pass
with pytest.raises(TypeError):
MCPServer("test", extensions=[_NoIdExt()])
class _VersionPinnedParams(types.RequestParams):
pass
class _VersionPinnedResult(types.Result):
ok: bool
class _VersionPinnedRequest(types.Request[_VersionPinnedParams, Literal["com.example/pinned"]]):
method: Literal["com.example/pinned"] = "com.example/pinned"
params: _VersionPinnedParams
class _VersionPinnedExt(Extension):
"""A method scoped to 2026-07-28 only via `MethodBinding.protocol_versions`."""
identifier = "com.example/pinned"
def methods(self):
async def handler(ctx: ServerRequestContext[Any, Any], params: _VersionPinnedParams) -> _VersionPinnedResult:
return _VersionPinnedResult(ok=True)
return [MethodBinding("com.example/pinned", _VersionPinnedParams, handler, frozenset({"2026-07-28"}))]
async def test_version_pinned_method_is_served_at_an_allowed_version() -> None:
"""SDK-defined: a `MethodBinding` with `protocol_versions` serves the method at a version
in the set."""
server = MCPServer("test", extensions=[_VersionPinnedExt()])
async with Client(server, mode="2026-07-28") as client:
request = _VersionPinnedRequest(params=_VersionPinnedParams())
result = await client.session.send_request(cast("types.ClientRequest", request), _VersionPinnedResult)
assert result == snapshot(_VersionPinnedResult(ok=True))
async def test_version_pinned_method_is_method_not_found_at_a_disallowed_version() -> None:
"""SDK-defined: the same method at a version outside `protocol_versions` is rejected with
METHOD_NOT_FOUND, mirroring the spec's per-version boundary."""
server = MCPServer("test", extensions=[_VersionPinnedExt()])
async with Client(server, mode="legacy") as client:
request = _VersionPinnedRequest(params=_VersionPinnedParams())
with pytest.raises(MCPError) as exc_info:
await client.session.send_request(cast("types.ClientRequest", request), _VersionPinnedResult)
assert exc_info.value.code == METHOD_NOT_FOUND
assert exc_info.value.error.data == "com.example/pinned"
@pytest.mark.parametrize(
"identifier",
[
"io.modelcontextprotocol/ui",
"com.example/my_ext",
"com.x-y.z2/n.a-b_c",
"example/x",
"a/b",
"com.example/9start",
],
)
def test_grammar_conformant_extension_identifiers_are_accepted(identifier: str) -> None:
"""Spec `_meta` key grammar: dot-separated labels (letter start, letter/digit end,
hyphens interior), a slash, then a name that starts and ends alphanumeric."""
validate_extension_identifier(identifier, owner="T")
@pytest.mark.parametrize(
"identifier",
[
"noprefix",
"-foo/bar",
".leading/x",
"a..b/x",
"foo-/x",
"9foo/x",
"foo/-bar",
"foo/bar-",
"foo/",
"/bar",
"foo/ba r",
"io.modelcontextprotocol/ui\n",
"",
None,
42,
],
)
def test_malformed_extension_identifiers_are_rejected(identifier: Any) -> None:
"""Spec `_meta` key grammar: malformed prefixes (bad label start/end, empty labels)
and malformed names are rejected, as are non-strings."""
with pytest.raises(TypeError):
validate_extension_identifier(identifier, owner="T")
@pytest.mark.parametrize("method", ["tools/list", "completion/complete"])
def test_method_binding_rejects_spec_methods(method: str) -> None:
"""SDK-defined: extension methods are additive — binding a spec-defined request method
would silently shadow (or be shadowed by) the server's own handler, so it is rejected
when the binding is constructed."""
with pytest.raises(ValueError):
MethodBinding(method, _PingParams, _pong_handler)
def test_method_binding_rejects_empty_protocol_versions() -> None:
"""SDK-defined: an empty `protocol_versions` set would make the method unreachable at
every version; `None` is the universal-version spelling."""
with pytest.raises(ValueError) as exc_info:
MethodBinding("com.example/dead", _PingParams, _pong_handler, frozenset())
assert str(exc_info.value) == snapshot(
"MethodBinding for 'com.example/dead' has an empty protocol_versions set, so it could "
"never be served; use None to admit every version"
)
class _OtherMethodExt(Extension):
"""A second extension binding the same verb as `_MethodExt`."""
identifier = "com.example/other-method"
def methods(self) -> list[MethodBinding]:
return [MethodBinding("com.example/ping", _PingParams, _pong_handler)]
def test_colliding_extension_methods_are_rejected_at_registration() -> None:
"""SDK-defined: two extensions binding the same method would silently last-write-win;
the collision is rejected when the second extension is applied."""
with pytest.raises(ValueError) as exc_info:
MCPServer("test", extensions=[_MethodExt(), _OtherMethodExt()])
assert str(exc_info.value) == snapshot(
"Extension 'com.example/other-method' binds method 'com.example/ping', which is already "
"registered; extension methods are additive and cannot replace another handler"
)
_NEEDS_EXT = "com.example/needed"
class _RequiresExt(Extension):
"""A tool that requires the client to have declared `com.example/needed`."""
identifier = _NEEDS_EXT
def tools(self):
def guarded(ctx: Context) -> str:
require_client_extension(ctx.request_context, _NEEDS_EXT)
return "ok"
return [ToolBinding(fn=guarded)]
async def test_require_client_extension_passes_when_client_declared_it() -> None:
"""SDK-defined: `require_client_extension` is a no-op when the client advertised the id."""
server = MCPServer("test", extensions=[_RequiresExt()])
async with Client(server, extensions={_NEEDS_EXT: {}}) as client:
result = await client.call_tool("guarded", {})
assert result == snapshot(CallToolResult(content=[TextContent(text="ok")], structured_content={"result": "ok"}))
async def test_require_client_extension_raises_minus_32021_when_client_did_not_declare_it() -> None:
"""SDK-defined: `require_client_extension` raises the -32021 missing-required-capability
error when the client did not advertise the id."""
server = MCPServer("test", extensions=[_RequiresExt()])
async with Client(server) as client:
with pytest.raises(MCPError) as exc_info:
await client.call_tool("guarded", {})
assert exc_info.value.code == MISSING_REQUIRED_CLIENT_CAPABILITY
assert exc_info.value.error.data == snapshot({"requiredCapabilities": {"extensions": {_NEEDS_EXT: {}}}})

302
tests/server/test_apps.py Normal file
View file

@ -0,0 +1,302 @@
"""Tests for the MCP Apps extension (`io.modelcontextprotocol/ui`, SEP-2133).
The headline property is SEP-2133 graceful degradation: a UI-bound tool returns
rich output to a client that negotiated Apps and text-only output to one that did
not. The remaining tests pin SDK-defined wiring (the `_meta.ui.resourceUri` stamp,
the `ui://` resource MIME type, capability advertisement, and `ui://`-scheme
validation).
"""
from typing import Any
import mcp_types as types
import pytest
from inline_snapshot import snapshot
from mcp_types import CallToolResult, ReadResourceResult, TextContent, TextResourceContents
from mcp.client.client import Client
from mcp.server import Server, ServerRequestContext
from mcp.server.apps import (
APP_MIME_TYPE,
EXTENSION_ID,
Apps,
ResourceCsp,
ResourcePermissions,
client_supports_apps,
)
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.context import Context
from mcp.server.mcpserver.resources import TextResource
pytestmark = pytest.mark.anyio
def _clock_server() -> MCPServer:
apps = Apps()
@apps.tool(resource_uri="ui://clock/app.html", title="Get Time", description="Return the current time.")
def get_time(ctx: Context) -> str:
if not client_supports_apps(ctx):
return "The time is 2026-06-26T00:00:00Z."
return "2026-06-26T00:00:00Z"
apps.add_html_resource("ui://clock/app.html", "<title>Clock</title>", title="Clock")
return MCPServer("clock", extensions=[apps])
async def test_apps_tool_stamps_ui_resource_uri_on_tool_meta() -> None:
"""SDK-defined: `@apps.tool(resource_uri=...)` stamps `_meta.ui.resourceUri` on the
advertised tool, observed end-to-end through `list_tools`."""
async with Client(_clock_server()) as client:
result = await client.list_tools()
assert [(t.name, t.meta) for t in result.tools] == snapshot(
[("get_time", {"ui": {"resourceUri": "ui://clock/app.html"}})]
)
async def test_add_html_resource_serves_ui_resource_at_app_mime_type() -> None:
"""SDK-defined: `add_html_resource` registers the `ui://` resource served as
`text/html;profile=mcp-app`, observed through `read_resource`."""
async with Client(_clock_server()) as client:
result = await client.read_resource("ui://clock/app.html")
assert result == snapshot(
ReadResourceResult(
contents=[
TextResourceContents(
uri="ui://clock/app.html",
mime_type="text/html;profile=mcp-app",
text="<title>Clock</title>",
)
]
)
)
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].mime_type == APP_MIME_TYPE
async def test_auto_mode_carries_apps_extension_under_server_capabilities() -> None:
"""SDK-defined: the Apps extension rides `server/discover`, so a `mode='auto'` client
sees `EXTENSION_ID` under `server_capabilities.extensions`."""
async with Client(_clock_server(), mode="auto") as client:
assert client.server_capabilities.extensions == snapshot({"io.modelcontextprotocol/ui": {}})
async def test_legacy_handshake_drops_apps_extension_from_capabilities() -> None:
"""Pinned gap: the 2025 `ServerCapabilities` wire schema has no `extensions` field,
so a `mode='legacy'` handshake cannot carry the Apps capability -- only `mode='auto'`
(server/discover) does. This pins the divergence rather than fixing it."""
async with Client(_clock_server(), mode="legacy") as client:
assert client.server_capabilities.extensions is None
async def test_apps_tool_returns_rich_output_when_client_negotiated_apps() -> None:
"""SEP-2133 graceful degradation: a client that advertised `EXTENSION_ID` gets the
rich (UI) path, while one that did not gets the text-only fallback. The same tool,
branching on `client_supports_apps(ctx)`, drives both halves."""
server = _clock_server()
async with Client(server, extensions={EXTENSION_ID: {"mimeTypes": [APP_MIME_TYPE]}}) as supports:
rich = await supports.call_tool("get_time", {})
async with Client(server) as plain:
fallback = await plain.call_tool("get_time", {})
assert rich.content == snapshot([TextContent(text="2026-06-26T00:00:00Z")])
assert fallback.content == snapshot([TextContent(text="The time is 2026-06-26T00:00:00Z.")])
async def _observed_client_supports_apps(extensions: dict[str, dict[str, Any]] | None) -> bool:
"""Run one probe `tools/call` and report what `client_supports_apps` saw server-side.
Exercises the lowlevel `ServerRequestContext` form, which reads the client's
advertised extensions off `session.client_params`.
"""
observed: list[bool] = []
async def list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(tools=[types.Tool(name="probe", input_schema={"type": "object"})])
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
assert params.name == "probe"
observed.append(client_supports_apps(ctx))
return CallToolResult(content=[TextContent(text="ok")])
server = Server("probe", on_list_tools=list_tools, on_call_tool=call_tool)
async with Client(server, extensions=extensions) as client:
await client.call_tool("probe", {})
return observed[0]
@pytest.mark.parametrize(
("extensions", "expected"),
[
pytest.param({EXTENSION_ID: {"mimeTypes": [APP_MIME_TYPE]}}, True, id="html-mime-listed"),
pytest.param({EXTENSION_ID: {"mimeTypes": (APP_MIME_TYPE,)}}, True, id="in-process-tuple-mime-types"),
pytest.param(None, False, id="extension-not-declared"),
pytest.param({EXTENSION_ID: {"mimeTypes": ["application/x-other"]}}, False, id="html-mime-not-offered"),
pytest.param({EXTENSION_ID: {}}, False, id="mime-types-key-missing"),
],
)
async def test_client_supports_apps_from_lowlevel_request_context(
extensions: dict[str, dict[str, Any]] | None, expected: bool
) -> None:
"""ext-apps: `client_supports_apps` is `True` only when the client declared the ui
extension AND listed `text/html;profile=mcp-app` in its `mimeTypes` settings — a
required field, so its absence means unsupported (the reference SDK's check is
`uiCap?.mimeTypes?.includes(...)`)."""
assert await _observed_client_supports_apps(extensions) is expected
def test_apps_tool_rejects_non_ui_resource_uri() -> None:
"""SDK-defined: `@apps.tool` accepts only `ui://` URIs; any other scheme is a
programmer error raised at decoration time."""
apps = Apps()
with pytest.raises(ValueError):
apps.tool(resource_uri="https://example.com/app.html")
def test_add_html_resource_rejects_non_ui_resource_uri() -> None:
"""SDK-defined: `add_html_resource` accepts only `ui://` URIs; any other scheme is
a programmer error raised at registration time."""
apps = Apps()
with pytest.raises(ValueError):
apps.add_html_resource("https://example.com/app.html", "<title>x</title>")
def _widget() -> str:
"""A UI-bound tool body (shared so its one covered call serves both meta tests)."""
return "x"
async def test_apps_tool_stamps_visibility_when_given() -> None:
"""SDK-defined: `@apps.tool(visibility=...)` is stamped into `_meta.ui.visibility`."""
apps = Apps()
apps.tool(resource_uri="ui://v/app.html", visibility=["app"])(_widget)
apps.add_html_resource("ui://v/app.html", "<title>v</title>")
async with Client(MCPServer("v", extensions=[apps])) as client:
result = await client.list_tools()
called = await client.call_tool("_widget", {})
assert result.tools[0].meta == snapshot({"ui": {"resourceUri": "ui://v/app.html", "visibility": ["app"]}})
assert called.content == snapshot([TextContent(text="x")])
async def test_apps_tool_merges_extra_meta_alongside_ui() -> None:
"""SDK-defined: `@apps.tool(meta=...)` merges extra `_meta` keys with the `ui` entry
(previously a `meta=` argument raised a duplicate-keyword TypeError)."""
apps = Apps()
apps.tool(resource_uri="ui://m/app.html", meta={"com.example/k": 1})(_widget)
apps.add_html_resource("ui://m/app.html", "<title>m</title>")
async with Client(MCPServer("m", extensions=[apps])) as client:
result = await client.list_tools()
assert result.tools[0].meta == snapshot({"com.example/k": 1, "ui": {"resourceUri": "ui://m/app.html"}})
async def test_add_html_resource_stamps_csp_and_permissions_on_resource_meta() -> None:
"""SDK-defined: `csp`/`permissions` populate the resource's `_meta.ui` per ext-apps."""
apps = Apps()
apps.add_html_resource(
"ui://r/app.html",
"<title>r</title>",
csp=ResourceCsp(connect_domains=["https://api.example.com"]),
permissions=ResourcePermissions(camera={}),
domain="r.example.com",
prefers_border=True,
)
async with Client(MCPServer("r", extensions=[apps])) as client:
listed = await client.list_resources()
result = await client.read_resource("ui://r/app.html")
expected_ui_meta = snapshot(
{
"ui": {
"csp": {"connectDomains": ["https://api.example.com"]},
"permissions": {"camera": {}},
"domain": "r.example.com",
"prefersBorder": True,
}
}
)
# Hosts read `_meta.ui` from the read content item, with the list entry as
# fallback — the SDK stamps the same value in both places.
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].meta == expected_ui_meta
assert listed.resources[0].meta == expected_ui_meta
def test_apps_tool_with_unregistered_resource_uri_is_rejected_at_construction() -> None:
"""SDK-defined: a tool whose `resource_uri` has no matching registered resource would
advertise a `_meta.ui.resourceUri` that 404s on `resources/read`; the misconfiguration
is rejected when the server consumes the extension."""
apps = Apps()
apps.tool(resource_uri="ui://missing/app.html")(_widget)
with pytest.raises(ValueError) as exc_info:
MCPServer("broken", extensions=[apps])
assert str(exc_info.value) == snapshot(
"Apps tool '_widget' binds resource_uri 'ui://missing/app.html', but no such resource "
"is registered; add it with add_html_resource() or add_resource()"
)
async def test_add_resource_registers_a_prebuilt_ui_resource() -> None:
"""SDK-defined: `add_resource` is the escape hatch for pre-built `ui://` resources
that `add_html_resource` cannot express; it satisfies a tool's `resource_uri` binding."""
apps = Apps()
apps.tool(resource_uri="ui://prebuilt/app.html")(_widget)
apps.add_resource(
TextResource(uri="ui://prebuilt/app.html", name="prebuilt", mime_type=APP_MIME_TYPE, text="<title>p</title>")
)
async with Client(MCPServer("p", extensions=[apps])) as client:
result = await client.read_resource("ui://prebuilt/app.html")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].text == "<title>p</title>"
def test_add_resource_rejects_non_ui_resource_uri() -> None:
"""SDK-defined: `add_resource` accepts only `ui://` URIs, like the other registrars."""
apps = Apps()
with pytest.raises(ValueError):
apps.add_resource(TextResource(uri="https://example.com/app.html", name="x", text="x"))
def test_apps_tool_rejects_a_ui_meta_key() -> None:
"""SDK-defined: the decorator owns `_meta['ui']` — a caller-supplied `'ui'` entry would be
silently clobbered, so it is rejected at decoration time (use `resource_uri=`/`visibility=`)."""
apps = Apps()
with pytest.raises(ValueError) as exc_info:
apps.tool(resource_uri="ui://c/app.html", meta={"ui": {"resourceUri": "ui://other.html"}})
assert str(exc_info.value) == snapshot(
"Apps.tool() owns _meta['ui']; pass resource_uri=/visibility= instead of a 'ui' meta key"
)
async def test_add_resource_defaults_the_mime_type_to_the_app_mime() -> None:
"""ext-apps: hosts only render `ui://` resources served as `text/html;profile=mcp-app`,
so a resource registered without an explicit `mime_type` gets it by default."""
apps = Apps()
apps.add_resource(TextResource(uri="ui://d/app.html", name="d", text="<title>d</title>"))
async with Client(MCPServer("d", extensions=[apps])) as client:
result = await client.read_resource("ui://d/app.html")
assert isinstance(result.contents[0], TextResourceContents)
assert result.contents[0].mime_type == APP_MIME_TYPE
def test_add_resource_rejects_an_explicit_non_app_mime_type() -> None:
"""ext-apps: an explicit `mime_type` other than `text/html;profile=mcp-app` would make
the resource unrenderable; the mismatch is rejected at registration."""
apps = Apps()
with pytest.raises(ValueError) as exc_info:
apps.add_resource(TextResource(uri="ui://e/app.html", name="e", mime_type="text/html", text="x"))
assert str(exc_info.value) == snapshot(
"MCP Apps resources are served as 'text/html;profile=mcp-app', got 'text/html'"
)

134
tests/server/test_extensions_capability.py Normal file
View file

@ -0,0 +1,134 @@
"""Tests for the SEP-2133 extensions capability negotiation plumbing.
The extension-map negotiation is independent of any concrete extension (Apps,
Tasks): the lowlevel `Server` advertises `self.extensions` under
`ServerCapabilities.extensions`, a client mirrors its own support under
`ClientCapabilities.extensions`, and `Connection.check_capability` resolves the
server-side query. These tests pin that plumbing end-to-end and at the unit
level. Per-extension contribution wiring lives in `test_extension.py`; this file
covers only the capability advertisement and negotiation.
"""
import mcp_types as types
import pytest
from inline_snapshot import snapshot
from mcp.client.client import Client
from mcp.server import Server, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import MCPServer
pytestmark = pytest.mark.anyio
_EXTENSION_ID = "com.example/x"
_OTHER_EXTENSION_ID = "com.example/other"
class _Extension(Extension):
identifier = _EXTENSION_ID
def settings(self) -> dict[str, object]:
return {"k": 1}
def test_get_capabilities_omits_extensions_when_none_registered() -> None:
"""SDK-defined: a lowlevel `Server` with an empty `extensions` map advertises
`ServerCapabilities.extensions` as `None`, not an empty map."""
server = Server("bare")
assert server.get_capabilities().extensions is None
def test_get_capabilities_advertises_populated_self_extensions() -> None:
"""SDK-defined: `get_capabilities` reads `self.extensions` (the map higher
layers populate) and advertises it under `ServerCapabilities.extensions`."""
server = Server("with-ext")
settings = {"k": 1}
server.extensions = {_EXTENSION_ID: settings}
assert server.get_capabilities().extensions == {_EXTENSION_ID: settings}
async def test_modern_connection_carries_the_advertised_extensions_map() -> None:
"""SDK-defined: over a modern (`server/discover`) connection the client reads
the server's advertised extension map from `server_capabilities`."""
server = MCPServer("host", extensions=[_Extension()])
async with Client(server, mode="auto") as client:
assert client.server_capabilities.extensions == snapshot({"com.example/x": {"k": 1}})
async def test_legacy_handshake_drops_the_extensions_map() -> None:
"""Pinned gap: the handshake-era `initialize` result is serialized against the
2025 wire schema, which has no `extensions` field, so a legacy handshake cannot
carry it; the client sees `None` even though the server advertised one."""
server = MCPServer("host", extensions=[_Extension()])
async with Client(server, mode="legacy") as client:
assert client.server_capabilities.extensions is None
async def test_server_accepts_capability_for_client_advertised_extension() -> None:
"""SDK-defined: a client advertising `extensions={id: ...}` makes the
server-side `check_client_capability` return True when queried for that id.
Observed inside a tool handler."""
queried = types.ClientCapabilities(extensions={_EXTENSION_ID: {}})
supported: list[bool] = []
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
assert params.name == "probe"
supported.append(ctx.session.check_client_capability(queried))
return types.CallToolResult(content=[])
async def list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(tools=[types.Tool(name="probe", input_schema={"type": "object"})])
server = Server("checker", on_call_tool=call_tool, on_list_tools=list_tools)
async with Client(server, extensions={_EXTENSION_ID: {"mimeTypes": ["text/html"]}}) as client:
await client.call_tool("probe", {})
assert supported == [True]
async def test_server_rejects_capability_for_undeclared_extension() -> None:
"""SDK-defined: when the client advertises one extension, a server query for a
*different* identifier returns False - presence, not value, is the check."""
queried = types.ClientCapabilities(extensions={_OTHER_EXTENSION_ID: {}})
supported: list[bool] = []
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
assert params.name == "probe"
supported.append(ctx.session.check_client_capability(queried))
return types.CallToolResult(content=[])
async def list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(tools=[types.Tool(name="probe", input_schema={"type": "object"})])
server = Server("checker", on_call_tool=call_tool, on_list_tools=list_tools)
async with Client(server, extensions={_EXTENSION_ID: {"mimeTypes": ["text/html"]}}) as client:
await client.call_tool("probe", {})
assert supported == [False]
async def test_server_rejects_capability_when_client_advertises_no_extensions() -> None:
"""SDK-defined: a client that declares no extensions makes any server
`check_client_capability` query for an extension return False."""
queried = types.ClientCapabilities(extensions={_EXTENSION_ID: {}})
supported: list[bool] = []
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
assert params.name == "probe"
supported.append(ctx.session.check_client_capability(queried))
return types.CallToolResult(content=[])
async def list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(tools=[types.Tool(name="probe", input_schema={"type": "object"})])
server = Server("checker", on_call_tool=call_tool, on_list_tools=list_tools)
async with Client(server) as client:
await client.call_tool("probe", {})
assert supported == [False]