Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/client/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,8 @@ That is why `main` narrows with `isinstance(block, TextContent)` before touching

When both are present they say the same thing twice on purpose: `content` is for a model, `structured_content` is for code. Where the structured half comes from, and how to control it, is the **[Structured Output](../servers/structured-output.md)** page.

The client validates `structured_content` against schemas learned from `list_tools()`. If a server exposes tools through a search or catalog API instead of listing them all, call `register_tool_schema(name, output_schema)` (on `Client` or `ClientSession`) before `call_tool` so those results are validated the same way. A later complete `list_tools()` that omits that name drops the registration — re-register if you still need it.

### `is_error`: whether the tool failed

A tool that raises does **not** raise in your client. It comes back as an ordinary result with `is_error=True`.
Expand Down
8 changes: 8 additions & 0 deletions src/mcp/client/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -943,6 +943,14 @@ async def list_tools(
),
)

def register_tool_schema(self, name: str, output_schema: dict[str, Any] | None = None) -> None:
"""Register a tool's output schema for result validation.

Delegates to `ClientSession.register_tool_schema`. Use when tools are discovered
dynamically and will not appear in `list_tools()` responses.
"""
self.session.register_tool_schema(name, output_schema)

@deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def send_roots_list_changed(self) -> None:
"""Send a notification that the roots list has changed."""
Expand Down
33 changes: 29 additions & 4 deletions src/mcp/client/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -405,8 +405,8 @@ def __init__(
self._log_level: types.LoggingLevel | None = log_level
self._message_handler = message_handler or _default_message_handler
self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
# Compiled output-schema validators, derived from `_tool_output_schemas` and owned by
# `_absorb_tool_listing`, which evicts a tool's entry whenever its schema changes.
# Compiled output-schema validators, derived from `_tool_output_schemas`. Evicted by
# `_absorb_tool_listing` and `register_tool_schema` whenever a tool's schema changes.
self._tool_output_validators: dict[str, Validator] = {}
self._x_mcp_header_maps: dict[str, dict[tuple[str, ...], str]] = {}
self._initialize_result: types.InitializeResult | None = None
Expand Down Expand Up @@ -1077,6 +1077,30 @@ def _resolve_param_headers(self, name: str, arguments: Mapping[str, Any]) -> dic
return {}
return mcp_param_headers(header_map, arguments)

def register_tool_schema(self, name: str, output_schema: dict[str, Any] | None = None) -> None:
"""Register a tool's output schema for result validation.

Use this when tools are discovered dynamically (for example via a catalog search API)
and will not appear in `list_tools()` responses. Writes into the same cache that
`list_tools()` / `_absorb_tool_listing` populate, and evicts any compiled validator
when the registered schema differs from the previously cached one.

A later complete (uncursored, single-page) `list_tools()` that omits `name` drops the
registration, the same prune path used for listing-absorbed schemas. Re-register after
such a listing if the tool is still in use. If the listing includes `name`, the listed
`outputSchema` replaces this registration.

Args:
name: Tool name as passed to `call_tool`.
output_schema: JSON Schema for `structuredContent`, or `None` when the tool has no
output schema (suppresses the "not listed" warning without validating).
"""
if name in self._tool_output_validators and not _same_schema(
self._tool_output_schemas.get(name), output_schema
):
del self._tool_output_validators[name]
self._tool_output_schemas[name] = output_schema

async def validate_tool_result(self, name: str, result: types.CallToolResult) -> None:
"""Revalidate a `CallToolResult` against the tool's declared output schema.

Expand Down Expand Up @@ -1114,8 +1138,9 @@ def _output_schema_validator(self, name: str, output_schema: dict[str, Any]) ->

Compiling is ~60x the cost of validating, so a one-shot `jsonschema.validate()` per
result dominates `call_tool`; the compiled validator is cached instead. It stays valid
because `_absorb_tool_listing` evicts a tool's validator whenever it absorbs a different
schema for that tool, so a cached entry always matches `output_schema`.
because `_absorb_tool_listing` and `register_tool_schema` evict a tool's validator
whenever they store a different schema for that tool, so a cached entry always matches
`output_schema`.

Raises:
RuntimeError: The schema is not a valid JSON Schema. Raised on every call, since a
Expand Down
156 changes: 156 additions & 0 deletions tests/client/test_register_tool_schema.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,156 @@
"""`ClientSession.register_tool_schema` for tools absent from `list_tools`."""

import logging

import pytest
from mcp_types import (
CallToolRequestParams,
CallToolResult,
ListToolsResult,
PaginatedRequestParams,
Tool,
)

from mcp.client.client import Client
from mcp.server import Server, ServerRequestContext

_SCORE_SCHEMA: dict[str, object] = {
"type": "object",
"properties": {"score": {"type": "integer"}},
"required": ["score"],
}
_SCORE_AS_STRING_SCHEMA: dict[str, object] = {
"type": "object",
"properties": {"score": {"type": "string"}},
"required": ["score"],
}


def _dynamic_tool_server(*, structured_content: dict[str, object]) -> Server:
"""`list_tools` advertises only a search meta-tool; `analyze` is callable but unlisted."""

async def on_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(tools=[Tool(name="search", input_schema={"type": "object"})])

async def on_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
assert params.name == "analyze"
return CallToolResult(content=[], structured_content=structured_content)

return Server("test-server", on_list_tools=on_list_tools, on_call_tool=on_call_tool)


@pytest.mark.anyio
async def test_register_tool_schema_lets_call_tool_validate_an_unlisted_tool() -> None:
"""SDK-defined: a schema registered for a tool absent from list_tools is used by call_tool."""
server = _dynamic_tool_server(structured_content={"score": 1})
async with Client(server) as client:
client.register_tool_schema("analyze", _SCORE_SCHEMA)
result = await client.call_tool("analyze", {})
assert result.structured_content == {"score": 1}


@pytest.mark.anyio
async def test_register_tool_schema_makes_call_tool_reject_nonconforming_structured_content() -> None:
"""Without registration, an unlisted tool skips validation; with it, mismatches raise."""
server = _dynamic_tool_server(structured_content={"score": "no"})
async with Client(server) as client:
# Unregistered: validation is skipped (tool never appears in list_tools).
skipped = await client.call_tool("analyze", {})
assert skipped.structured_content == {"score": "no"}

client.register_tool_schema("analyze", _SCORE_SCHEMA)
# Stable SDK prefix only: the message tail is jsonschema text that shifts with the dependency.
with pytest.raises(RuntimeError, match="Invalid structured content returned by tool analyze"):
await client.call_tool("analyze", {})


@pytest.mark.anyio
async def test_register_tool_schema_with_none_suppresses_the_unlisted_warning(
caplog: pytest.LogCaptureFixture,
) -> None:
"""SDK-defined: registering None marks the tool known without validating structuredContent."""
server = _dynamic_tool_server(structured_content={"anything": True})
async with Client(server) as client:
client.register_tool_schema("analyze", None)
with caplog.at_level(logging.WARNING, logger="client"):
result = await client.call_tool("analyze", {})
assert result.structured_content == {"anything": True}
assert "not listed by server" not in caplog.text


@pytest.mark.anyio
async def test_register_tool_schema_evicts_the_compiled_validator_when_the_schema_changes() -> None:
"""SDK-defined: a changed registration must not reuse a validator compiled for the old schema."""
server = _dynamic_tool_server(structured_content={"score": 1})
async with Client(server) as client:
client.register_tool_schema("analyze", _SCORE_SCHEMA)
await client.session.validate_tool_result(
"analyze", CallToolResult(content=[], structured_content={"score": 1})
)
compiled = client.session._tool_output_validators["analyze"]

client.register_tool_schema("analyze", _SCORE_AS_STRING_SCHEMA)
assert "analyze" not in client.session._tool_output_validators

with pytest.raises(RuntimeError, match="Invalid structured content returned by tool analyze"):
await client.session.validate_tool_result(
"analyze", CallToolResult(content=[], structured_content={"score": 1})
)
assert client.session._tool_output_validators["analyze"] is not compiled


@pytest.mark.anyio
async def test_register_tool_schema_keeps_the_validator_when_the_schema_is_unchanged() -> None:
"""SDK-defined: re-registering an equal schema keeps the compiled validator."""
server = _dynamic_tool_server(structured_content={"score": 1})
async with Client(server) as client:
client.register_tool_schema("analyze", _SCORE_SCHEMA)
result = CallToolResult(content=[], structured_content={"score": 1})
await client.session.validate_tool_result("analyze", result)
compiled = client.session._tool_output_validators["analyze"]

client.register_tool_schema("analyze", dict(_SCORE_SCHEMA))
await client.session.validate_tool_result("analyze", result)
assert client.session._tool_output_validators["analyze"] is compiled


@pytest.mark.anyio
async def test_a_complete_list_tools_prunes_a_manually_registered_schema() -> None:
"""SDK-defined: a complete listing is still the full tool universe for prune — a registered
tool omitted from that listing is dropped, same as listing-absorbed schemas."""
server = _dynamic_tool_server(structured_content={"score": 1})
async with Client(server) as client:
client.register_tool_schema("analyze", _SCORE_SCHEMA)
assert "analyze" in client.session._tool_output_schemas

await client.session.list_tools()
assert "analyze" not in client.session._tool_output_schemas
assert set(client.session._tool_output_schemas) == {"search"}


@pytest.mark.anyio
async def test_list_tools_that_includes_a_registered_name_replaces_the_registered_schema() -> None:
"""SDK-defined: when the same name later appears in a listing, the listed schema wins."""

async def on_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(
tools=[
Tool(name="analyze", input_schema={"type": "object"}, output_schema=_SCORE_AS_STRING_SCHEMA),
]
)

async def on_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
raise NotImplementedError

server = Server("test-server", on_list_tools=on_list_tools, on_call_tool=on_call_tool)
async with Client(server) as client:
client.register_tool_schema("analyze", _SCORE_SCHEMA)
await client.session.validate_tool_result(
"analyze", CallToolResult(content=[], structured_content={"score": 1})
)

await client.session.list_tools()
with pytest.raises(RuntimeError, match="Invalid structured content returned by tool analyze"):
await client.session.validate_tool_result(
"analyze", CallToolResult(content=[], structured_content={"score": 1})
)
Loading