Max Isbey
a0d793154a
SEP-1686 makes a task-augmented `tools/call` answer with a `CreateTaskResult`,
but `SERVER_RESULTS[("tools/call", "2025-11-25")]` held `CallToolResult` alone.
The request half shipped and the response half did not, so a server returning
a task got `-32603 "Handler returned an invalid result"`, a server returning
both shapes had `task` silently sieved away, and a client receiving one raised
`ValidationError` before its caller saw anything. The v1.x line could express
that response; v2 cannot, on a revision the SDK still negotiates and still lets
a server advertise `capabilities.tasks` for.
Give the row its second arm as a generated `AnyCallToolResult`, alongside the
2026 aliases, and widen the lowlevel `Server`'s `on_call_tool` to match. Return
types are covariant, so a handler that only returns `CallToolResult` is
unaffected. Only the 2025-11-25 row gets the arm. The earlier handshake
revisions share `_v2025.CallToolRequest` and so already parse `params.task`, but
that is an artifact of the shared schema era, not a licence to answer: they
predate SEP-1686, and a clean INTERNAL_ERROR serves an off-spec client better
than a result shape its revision never defined. `MONOLITH_RESULTS["tools/call"]`
gains the arm too, so `parse_server_result` covers everything the surface admits.
`Task.ttl` is required and nullable ("null for unlimited"), and every dump path
passes `exclude_none=True`, which cannot tell a required null from an unset
optional and dropped it, leaving a body that fails the surface it was just
validated against. Rather than patch the 27 dump sites, models carrying such a
field now take a `KeepRequiredNullable` base that puts it back, and only that:
a field the caller filtered out with `include`/`exclude`, or one that was never
set, stays absent. The generator derives which classes need it from the schema
and a test asserts the same rule against the built models, so a nullable-required
field in a future revision is covered without anyone remembering. The field set
resolves on first dump rather than at class creation, because the generated
modules defer annotations and finish with `model_rebuild()`, so resolving early
would silently see nothing for a forward-referenced field.
That rule also reaches two live bugs of the same shape.
`LoggingMessageNotificationParams.data` is required and untyped, so
`ctx.log("info", None)` dropped the key and the receiving session rejected the
notification before dispatch: the message vanished with no error to either side,
at every protocol version. And `JSONRPCError.id` is required-and-nullable per
JSON-RPC 2.0, which the streamable-HTTP writer worked around by hand at one call
site; that patch is deleted and the base covers every writer.
The base is applied per model rather than to `MCPModel`/`WireModel`: a wrap
serializer costs per dump and pushes pydantic off its fast path for every model
in the tree, which measured 5x on a 20-tool `tools/list`. Its return is
deliberately unannotated, because an annotation there collapses the model's
serialization JSON schema to an opaque object.
The `tasks/*` lifecycle methods are left where they already work. They are
absent from `SPEC_CLIENT_METHODS`, so `Server.add_request_handler` serves them
today; adding registry rows would put those names into that version-flattened
set, which would make it illegal for an extension to bind `tasks/get` and would
gate the 2026 tasks extension's own calls to METHOD_NOT_FOUND. 2026-07-28 keeps
rejecting a `CreateTaskResult` on `tools/call`, which is correct: tasks left the
core protocol there.
569 lines
24 KiB
Python
569 lines
24 KiB
Python
"""Tool interactions against the low-level Server, driven through the public Client API."""
|
|
|
|
import anyio
|
|
import mcp_types as types
|
|
import pytest
|
|
from inline_snapshot import snapshot
|
|
from mcp_types import (
|
|
INVALID_PARAMS,
|
|
AudioContent,
|
|
CallToolResult,
|
|
EmbeddedResource,
|
|
ErrorData,
|
|
Icon,
|
|
ImageContent,
|
|
ListToolsResult,
|
|
ResourceLink,
|
|
TextContent,
|
|
TextResourceContents,
|
|
Tool,
|
|
ToolAnnotations,
|
|
)
|
|
from pydantic import TypeAdapter
|
|
|
|
from mcp import MCPError
|
|
from mcp.server import Server, ServerRequestContext
|
|
from tests._stamp import Unstamp
|
|
from tests.interaction._connect import Connect
|
|
from tests.interaction._requirements import requirement
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
@requirement("tools:call:content:text")
|
|
async def test_call_tool_returns_text_content(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""Arguments reach the tool handler; its content comes back as the call result."""
|
|
|
|
async def list_tools(
|
|
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
|
|
) -> types.ListToolsResult:
|
|
return types.ListToolsResult(
|
|
tools=[types.Tool(name="add", description="Add two integers.", input_schema={"type": "object"})]
|
|
)
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "add"
|
|
assert params.arguments is not None
|
|
return CallToolResult(content=[TextContent(text=str(params.arguments["a"] + params.arguments["b"]))])
|
|
|
|
server = Server("adder", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
result = await client.call_tool("add", {"a": 2, "b": 3})
|
|
|
|
assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="5")]))
|
|
|
|
|
|
@requirement("tools:call:is-error")
|
|
async def test_call_tool_execution_error_is_returned_as_result(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""A tool reporting its own failure with is_error=True reaches the client as a result, not an exception.
|
|
|
|
Tool execution errors are part of the result so the caller (typically a model) can see
|
|
them; only protocol-level failures become JSON-RPC errors.
|
|
"""
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "flux"
|
|
return CallToolResult(content=[TextContent(text="the flux capacitor is offline")], is_error=True)
|
|
|
|
server = Server("errors", on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
result = await client.call_tool("flux", {})
|
|
|
|
assert unstamped(result) == snapshot(
|
|
CallToolResult(content=[TextContent(text="the flux capacitor is offline")], is_error=True)
|
|
)
|
|
|
|
|
|
@requirement("tools:call:unknown-name")
|
|
async def test_call_tool_unknown_tool_is_protocol_error(connect: Connect) -> None:
|
|
"""A handler that rejects an unrecognised tool name with MCPError produces a JSON-RPC error.
|
|
|
|
The error's code, message, and data chosen by the handler reach the client verbatim.
|
|
"""
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
raise MCPError(code=INVALID_PARAMS, message=f"Unknown tool: {params.name}", data={"requested": params.name})
|
|
|
|
server = Server("errors", on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
with pytest.raises(MCPError) as exc_info:
|
|
await client.call_tool("nope", {})
|
|
|
|
assert exc_info.value.error == snapshot(
|
|
ErrorData(code=INVALID_PARAMS, message="Unknown tool: nope", data={"requested": "nope"})
|
|
)
|
|
|
|
|
|
@requirement("protocol:error:internal-error")
|
|
async def test_call_tool_uncaught_exception_becomes_error_response(connect: Connect) -> None:
|
|
"""An uncaught exception in the tool handler surfaces to the client as a JSON-RPC error.
|
|
|
|
The low-level server reports it with code 0 and the exception text as the message; see the
|
|
divergence note on the requirement.
|
|
"""
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "explode"
|
|
raise ValueError("boom")
|
|
|
|
server = Server("errors", on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
with pytest.raises(MCPError) as exc_info:
|
|
await client.call_tool("explode", {})
|
|
|
|
assert exc_info.value.error == snapshot(ErrorData(code=0, message="boom"))
|
|
|
|
|
|
@requirement("tools:list:basic")
|
|
async def test_list_tools_returns_registered_tools(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""The tools advertised by the server's list handler arrive at the client unchanged."""
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(
|
|
tools=[
|
|
Tool(
|
|
name="add",
|
|
description="Add two integers.",
|
|
input_schema={
|
|
"type": "object",
|
|
"properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
|
|
"required": ["a", "b"],
|
|
},
|
|
),
|
|
Tool(name="reset", description="Reset the calculator.", input_schema={"type": "object"}),
|
|
]
|
|
)
|
|
|
|
server = Server("calculator", on_list_tools=list_tools)
|
|
|
|
async with connect(server) as client:
|
|
result = await client.list_tools()
|
|
|
|
assert unstamped(result) == snapshot(
|
|
ListToolsResult(
|
|
tools=[
|
|
Tool(
|
|
name="add",
|
|
description="Add two integers.",
|
|
input_schema={
|
|
"type": "object",
|
|
"properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
|
|
"required": ["a", "b"],
|
|
},
|
|
),
|
|
Tool(name="reset", description="Reset the calculator.", input_schema={"type": "object"}),
|
|
]
|
|
)
|
|
)
|
|
|
|
|
|
@requirement("tools:input-schema:json-schema-2020-12")
|
|
@requirement("tools:input-schema:preserve-additional-properties")
|
|
@requirement("tools:input-schema:preserve-defs")
|
|
@requirement("tools:input-schema:preserve-schema-dialect")
|
|
async def test_tools_list_preserves_arbitrary_input_schema_keywords(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""A rich JSON Schema 2020-12 inputSchema reaches the client unchanged and the tool is callable.
|
|
|
|
The single identity assertion below proves all four pass-through behaviours at once: the same
|
|
dict literal that was registered is the dict that arrives, so $schema, $defs, the nested object
|
|
property, and additionalProperties are each preserved by virtue of the whole schema being
|
|
preserved. The follow-up call proves the rich-schema tool is callable end to end.
|
|
"""
|
|
schema = {
|
|
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
|
"type": "object",
|
|
"$defs": {"positive": {"type": "integer", "exclusiveMinimum": 0}},
|
|
"properties": {
|
|
"count": {"$ref": "#/$defs/positive"},
|
|
"options": {
|
|
"type": "object",
|
|
"properties": {"verbose": {"type": "boolean"}},
|
|
"additionalProperties": False,
|
|
},
|
|
},
|
|
"required": ["count"],
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(tools=[Tool(name="typed", input_schema=schema)])
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "typed"
|
|
assert params.arguments == {"count": 3, "options": {"verbose": True}}
|
|
return CallToolResult(content=[TextContent(text="ok")])
|
|
|
|
server = Server("typed", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
listed = await client.list_tools()
|
|
called = await client.call_tool("typed", {"count": 3, "options": {"verbose": True}})
|
|
|
|
assert listed.tools[0].input_schema == schema
|
|
assert unstamped(called) == snapshot(CallToolResult(content=[TextContent(text="ok")]))
|
|
|
|
|
|
@requirement("tools:list:metadata")
|
|
async def test_list_tools_optional_fields_round_trip(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""Every optional Tool field the server supplies reaches the client unchanged."""
|
|
|
|
tool = Tool(
|
|
name="annotated",
|
|
title="Annotated tool",
|
|
description="A tool carrying every optional field.",
|
|
input_schema={"type": "object"},
|
|
output_schema={"type": "object", "properties": {"answer": {"type": "integer"}}},
|
|
icons=[Icon(src="https://example.com/icon.png", mime_type="image/png", sizes=["48x48"])],
|
|
annotations=ToolAnnotations(title="Display title", read_only_hint=True, idempotent_hint=True),
|
|
_meta={"example.com/source": "interaction-suite"},
|
|
)
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(tools=[tool])
|
|
|
|
server = Server("annotated", on_list_tools=list_tools)
|
|
|
|
async with connect(server) as client:
|
|
result = await client.list_tools()
|
|
|
|
assert unstamped(result) == snapshot(
|
|
ListToolsResult(
|
|
tools=[
|
|
Tool(
|
|
name="annotated",
|
|
title="Annotated tool",
|
|
description="A tool carrying every optional field.",
|
|
input_schema={"type": "object"},
|
|
output_schema={"type": "object", "properties": {"answer": {"type": "integer"}}},
|
|
icons=[Icon(src="https://example.com/icon.png", mime_type="image/png", sizes=["48x48"])],
|
|
annotations=ToolAnnotations(title="Display title", read_only_hint=True, idempotent_hint=True),
|
|
_meta={"example.com/source": "interaction-suite"},
|
|
)
|
|
]
|
|
)
|
|
)
|
|
|
|
|
|
@requirement("tools:call:content:mixed")
|
|
@requirement("tools:call:content:image")
|
|
@requirement("tools:call:content:audio")
|
|
@requirement("tools:call:content:resource-link")
|
|
@requirement("tools:call:content:embedded-resource")
|
|
async def test_call_tool_multiple_content_block_types(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""A tool result can mix every content block type; all of them arrive in order.
|
|
|
|
The payloads are tiny fixed base64 strings ("aW1n" is b"img", "YXVk" is b"aud") so the
|
|
snapshot pins the exact bytes the client receives.
|
|
"""
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(tools=[Tool(name="render", input_schema={"type": "object"})])
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "render"
|
|
return CallToolResult(
|
|
content=[
|
|
TextContent(text="all five content block types"),
|
|
ImageContent(data="aW1n", mime_type="image/png"),
|
|
AudioContent(data="YXVk", mime_type="audio/wav"),
|
|
ResourceLink(name="report", uri="resource://reports/1", description="The full report"),
|
|
EmbeddedResource(
|
|
resource=TextResourceContents(uri="resource://reports/1", mime_type="text/plain", text="contents")
|
|
),
|
|
]
|
|
)
|
|
|
|
server = Server("renderer", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
result = await client.call_tool("render", {})
|
|
|
|
assert unstamped(result) == snapshot(
|
|
CallToolResult(
|
|
content=[
|
|
TextContent(text="all five content block types"),
|
|
ImageContent(data="aW1n", mime_type="image/png"),
|
|
AudioContent(data="YXVk", mime_type="audio/wav"),
|
|
ResourceLink(name="report", uri="resource://reports/1", description="The full report"),
|
|
EmbeddedResource(
|
|
resource=TextResourceContents(uri="resource://reports/1", mime_type="text/plain", text="contents")
|
|
),
|
|
]
|
|
)
|
|
)
|
|
|
|
|
|
@requirement("tools:call:structured-content")
|
|
async def test_call_tool_structured_content(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""A tool result carrying structured content alongside content delivers both to the client."""
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(tools=[Tool(name="sum", input_schema={"type": "object"})])
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "sum"
|
|
return CallToolResult(content=[TextContent(text="the sum is 5")], structured_content={"sum": 5})
|
|
|
|
server = Server("calculator", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
result = await client.call_tool("sum", {})
|
|
|
|
assert unstamped(result) == snapshot(
|
|
CallToolResult(content=[TextContent(text="the sum is 5")], structured_content={"sum": 5})
|
|
)
|
|
|
|
|
|
@requirement("tools:call:concurrent")
|
|
async def test_concurrent_tool_calls_complete_independently(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""Two tool calls in flight at once run concurrently and each caller gets its own answer.
|
|
|
|
Both handlers are held on a shared event after signalling that they have started, and the test
|
|
only releases them once both signals have arrived -- a server that processed requests
|
|
sequentially would never start the second handler and the test would time out instead.
|
|
"""
|
|
started: list[str] = []
|
|
started_events = {"first": anyio.Event(), "second": anyio.Event()}
|
|
release = anyio.Event()
|
|
results: dict[str, CallToolResult] = {}
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(tools=[Tool(name="echo", input_schema={"type": "object"})])
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "echo"
|
|
assert params.arguments is not None
|
|
tag = params.arguments["tag"]
|
|
assert isinstance(tag, str)
|
|
started.append(tag)
|
|
started_events[tag].set()
|
|
await release.wait()
|
|
return CallToolResult(content=[TextContent(text=tag)])
|
|
|
|
server = Server("echoer", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
with anyio.fail_after(5):
|
|
async with anyio.create_task_group() as task_group: # pragma: no branch
|
|
|
|
async def call_and_record(tag: str) -> None:
|
|
results[tag] = unstamped(await client.call_tool("echo", {"tag": tag}))
|
|
|
|
task_group.start_soon(call_and_record, "first")
|
|
task_group.start_soon(call_and_record, "second")
|
|
|
|
# Both handlers are running at the same time before either is allowed to finish.
|
|
await started_events["first"].wait()
|
|
await started_events["second"].wait()
|
|
release.set()
|
|
|
|
assert sorted(started) == ["first", "second"]
|
|
assert results == snapshot(
|
|
{
|
|
"first": CallToolResult(content=[TextContent(text="first")]),
|
|
"second": CallToolResult(content=[TextContent(text="second")]),
|
|
}
|
|
)
|
|
|
|
|
|
@requirement("client:output-schema:validate")
|
|
async def test_call_tool_structured_content_violating_output_schema_is_rejected_by_the_client(connect: Connect) -> None:
|
|
"""A result whose structured content does not conform to the tool's declared output schema never
|
|
reaches the caller: the client validates it against the schema cached from tools/list and raises.
|
|
"""
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(
|
|
tools=[
|
|
Tool(
|
|
name="forecast",
|
|
input_schema={"type": "object"},
|
|
output_schema={
|
|
"type": "object",
|
|
"properties": {"temperature": {"type": "number"}},
|
|
"required": ["temperature"],
|
|
},
|
|
)
|
|
]
|
|
)
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "forecast"
|
|
return CallToolResult(content=[TextContent(text="warm")], structured_content={"temperature": "warm"})
|
|
|
|
server = Server("weather", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
await client.list_tools()
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
await client.call_tool("forecast", {})
|
|
|
|
# The message embeds the jsonschema validation error, so only the SDK-authored prefix is pinned.
|
|
assert str(exc_info.value).startswith("Invalid structured content returned by tool forecast")
|
|
|
|
|
|
@requirement("client:output-schema:skip-on-error")
|
|
async def test_is_error_result_bypasses_client_output_schema_validation(connect: Connect, unstamped: Unstamp) -> None:
|
|
"""A tool result with isError true is returned as-is even when its structured content violates the schema.
|
|
|
|
The schema is cached up front so the client could validate, proving the bypass is specifically the
|
|
isError flag and not an empty cache.
|
|
"""
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(
|
|
tools=[
|
|
Tool(
|
|
name="forecast",
|
|
input_schema={"type": "object"},
|
|
output_schema={
|
|
"type": "object",
|
|
"properties": {"temperature": {"type": "number"}},
|
|
"required": ["temperature"],
|
|
},
|
|
)
|
|
]
|
|
)
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "forecast"
|
|
return CallToolResult(
|
|
content=[TextContent(text="boom")], structured_content={"temperature": "warm"}, is_error=True
|
|
)
|
|
|
|
server = Server("weather", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
await client.list_tools()
|
|
result = await client.call_tool("forecast", {})
|
|
|
|
assert unstamped(result) == snapshot(
|
|
CallToolResult(content=[TextContent(text="boom")], structured_content={"temperature": "warm"}, is_error=True)
|
|
)
|
|
|
|
|
|
@requirement("client:output-schema:missing-structured")
|
|
async def test_declared_output_schema_with_no_structured_content_is_rejected_by_the_client(connect: Connect) -> None:
|
|
"""A tool that declared an output schema but returned no structuredContent fails the client-side check.
|
|
|
|
The error is the SDK's own message, so the full text is snapshotted.
|
|
"""
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
return ListToolsResult(
|
|
tools=[
|
|
Tool(
|
|
name="forecast",
|
|
input_schema={"type": "object"},
|
|
output_schema={"type": "object", "properties": {"temperature": {"type": "number"}}},
|
|
)
|
|
]
|
|
)
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "forecast"
|
|
return CallToolResult(content=[TextContent(text="warm")])
|
|
|
|
server = Server("weather", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
await client.list_tools()
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
await client.call_tool("forecast", {})
|
|
|
|
assert str(exc_info.value) == snapshot("Tool forecast has an output schema but did not return structured content")
|
|
|
|
|
|
@requirement("client:output-schema:auto-list")
|
|
async def test_call_tool_populates_the_output_schema_cache_via_an_implicit_tools_list(
|
|
connect: Connect, unstamped: Unstamp
|
|
) -> None:
|
|
"""Calling a tool whose schema is not cached issues exactly one implicit tools/list to populate it.
|
|
|
|
The first call_tool of an uncached tool triggers a tools/list the caller never asked for; the
|
|
second call hits the cache and does not. This is the SDK's chosen cache strategy and the cause of
|
|
the surprising behaviour where a server with only on_call_tool sees a successful call answered
|
|
with METHOD_NOT_FOUND from a request the caller never made; see the divergence on the requirement.
|
|
"""
|
|
list_calls: list[str] = []
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
list_calls.append("called")
|
|
return ListToolsResult(
|
|
tools=[
|
|
Tool(
|
|
name="forecast",
|
|
input_schema={"type": "object"},
|
|
output_schema={"type": "object", "properties": {"temperature": {"type": "number"}}},
|
|
)
|
|
]
|
|
)
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "forecast"
|
|
return CallToolResult(content=[TextContent(text="21 C")], structured_content={"temperature": 21})
|
|
|
|
server = Server("weather", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
first = await client.call_tool("forecast", {})
|
|
assert list_calls == ["called"]
|
|
second = await client.call_tool("forecast", {})
|
|
|
|
assert list_calls == ["called"]
|
|
assert unstamped(first) == snapshot(
|
|
CallToolResult(content=[TextContent(text="21 C")], structured_content={"temperature": 21})
|
|
)
|
|
assert unstamped(second) == first
|
|
|
|
|
|
@requirement("tools:call:task-augmented")
|
|
async def test_task_augmented_call_tool_may_be_answered_with_a_create_task_result(connect: Connect) -> None:
|
|
"""A 2025-11-25 server can hand back a task instead of running the tool inline.
|
|
|
|
SEP-1686 makes `task` an opt-in on the request and `CreateTaskResult` the answer to it. The
|
|
SDK carries the vocabulary and validates both halves; the task store and the `tasks/*`
|
|
lifecycle handlers belong to the server author, so this drives the create step only. The
|
|
caller names the result type because `call_tool` resolves to the two core arms.
|
|
"""
|
|
task = types.Task(
|
|
task_id="task-1",
|
|
status="working",
|
|
created_at="2026-07-24T00:00:00Z",
|
|
last_updated_at="2026-07-24T00:00:00Z",
|
|
ttl=None,
|
|
)
|
|
|
|
async def call_tool(
|
|
ctx: ServerRequestContext, params: types.CallToolRequestParams
|
|
) -> CallToolResult | types.CreateTaskResult:
|
|
assert params.name == "render"
|
|
if params.task is not None:
|
|
assert params.task.ttl == 60_000
|
|
return types.CreateTaskResult(task=task)
|
|
return CallToolResult(content=[TextContent(text="rendered inline")])
|
|
|
|
server = Server("renderer", on_call_tool=call_tool)
|
|
result_type = TypeAdapter[CallToolResult | types.CreateTaskResult](CallToolResult | types.CreateTaskResult)
|
|
|
|
async with connect(server) as client:
|
|
tasked = await client.session.send_request(
|
|
types.CallToolRequest(
|
|
params=types.CallToolRequestParams(name="render", task=types.TaskMetadata(ttl=60_000))
|
|
),
|
|
result_type,
|
|
)
|
|
inline = await client.session.send_request(
|
|
types.CallToolRequest(params=types.CallToolRequestParams(name="render")), result_type
|
|
)
|
|
|
|
assert isinstance(tasked, types.CreateTaskResult)
|
|
# Whole-object equality: `ttl` is required and nullable, and a dump that drops it
|
|
# leaves a body the surface would reject, so the null has to survive the round trip.
|
|
assert tasked.task == task
|
|
assert isinstance(inline, CallToolResult)
|