Max Isbey
1b74b06753
Cut comment and docstring volume roughly in half across src, tests, examples, and docs_src: removed comments that restate the adjacent code, leftover development narration, section banners, and self-evident Args/Returns blocks, and compressed the remaining docstrings to a Google-style summary line plus only the detail that earns its place. Kept (and tightened) the load-bearing content: Raises sections, deprecation and version-availability notes, spec/RFC/issue references, why-comments for non-obvious decisions, and all coverage pragmas. The generated mcp_types.v* wire modules are untouched.
346 lines
14 KiB
Python
346 lines
14 KiB
Python
"""Initialization handshake against the low-level Server, driven through the public Client API.
|
|
|
|
Client always performs the full handshake at the latest protocol version, so tests that skip
|
|
initialization or vary the requested version drive a bare ClientSession over an InMemoryTransport.
|
|
The last two tests script the server's side of the wire by hand — a pattern reserved for behaviour
|
|
no real Server can be made to produce.
|
|
"""
|
|
|
|
import anyio
|
|
import mcp_types as types
|
|
import pytest
|
|
from inline_snapshot import snapshot
|
|
from mcp_types import (
|
|
INVALID_PARAMS,
|
|
CallToolResult,
|
|
ClientCapabilities,
|
|
CompletionsCapability,
|
|
EmptyResult,
|
|
ErrorData,
|
|
Icon,
|
|
Implementation,
|
|
InitializeRequest,
|
|
InitializeRequestParams,
|
|
InitializeResult,
|
|
JSONRPCRequest,
|
|
JSONRPCResponse,
|
|
ListToolsRequest,
|
|
ListToolsResult,
|
|
LoggingCapability,
|
|
PromptsCapability,
|
|
ResourcesCapability,
|
|
ServerCapabilities,
|
|
TextContent,
|
|
ToolsCapability,
|
|
)
|
|
|
|
from mcp import MCPError
|
|
from mcp.client import ClientRequestContext, ClientSession
|
|
from mcp.client._memory import InMemoryTransport
|
|
from mcp.server import Server, ServerRequestContext
|
|
from mcp.shared.memory import MessageStream, create_client_server_memory_streams
|
|
from mcp.shared.message import SessionMessage
|
|
from tests.interaction._connect import Connect
|
|
from tests.interaction._requirements import requirement
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
@requirement("lifecycle:initialize:basic")
|
|
@requirement("lifecycle:initialize:server-info")
|
|
async def test_initialize_returns_server_info(connect: Connect) -> None:
|
|
server = Server(
|
|
"greeter",
|
|
version="1.2.3",
|
|
title="Greeter",
|
|
description="Greets people.",
|
|
website_url="https://example.com/greeter",
|
|
icons=[Icon(src="https://example.com/icon.png", mime_type="image/png", sizes=["48x48"])],
|
|
)
|
|
|
|
async with connect(server) as client:
|
|
server_info = client.server_info
|
|
|
|
assert server_info == snapshot(
|
|
Implementation(
|
|
name="greeter",
|
|
title="Greeter",
|
|
description="Greets people.",
|
|
version="1.2.3",
|
|
website_url="https://example.com/greeter",
|
|
icons=[Icon(src="https://example.com/icon.png", mime_type="image/png", sizes=["48x48"])],
|
|
)
|
|
)
|
|
|
|
|
|
@requirement("lifecycle:initialize:instructions")
|
|
async def test_initialize_returns_instructions(connect: Connect) -> None:
|
|
async with connect(Server("guided", instructions="Call the add tool.")) as client:
|
|
assert client.instructions == snapshot("Call the add tool.")
|
|
|
|
async with connect(Server("unguided")) as client:
|
|
assert client.instructions is None
|
|
|
|
|
|
@requirement("lifecycle:initialize:capabilities:from-handlers")
|
|
@requirement("tools:capability:declared")
|
|
@requirement("resources:capability:declared")
|
|
@requirement("prompts:capability:declared")
|
|
@requirement("completion:capability:declared")
|
|
async def test_initialize_capabilities_reflect_registered_handlers(connect: Connect) -> None:
|
|
"""The in-memory transport connects with default initialization options, so list_changed is always False."""
|
|
|
|
async def list_tools(
|
|
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
|
|
) -> types.ListToolsResult:
|
|
raise NotImplementedError
|
|
|
|
async def list_resources(
|
|
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
|
|
) -> types.ListResourcesResult:
|
|
raise NotImplementedError
|
|
|
|
async def subscribe_resource(ctx: ServerRequestContext, params: types.SubscribeRequestParams) -> types.EmptyResult:
|
|
raise NotImplementedError
|
|
|
|
async def list_prompts(
|
|
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
|
|
) -> types.ListPromptsResult:
|
|
raise NotImplementedError
|
|
|
|
async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelRequestParams) -> types.EmptyResult:
|
|
raise NotImplementedError
|
|
|
|
async def completion(ctx: ServerRequestContext, params: types.CompleteRequestParams) -> types.CompleteResult:
|
|
raise NotImplementedError
|
|
|
|
server = Server( # pyright: ignore[reportDeprecated]
|
|
"full",
|
|
on_list_tools=list_tools,
|
|
on_list_resources=list_resources,
|
|
on_subscribe_resource=subscribe_resource,
|
|
on_list_prompts=list_prompts,
|
|
on_set_logging_level=set_logging_level,
|
|
on_completion=completion,
|
|
)
|
|
|
|
async with connect(server) as client:
|
|
capabilities = client.server_capabilities
|
|
|
|
assert capabilities == snapshot(
|
|
ServerCapabilities(
|
|
experimental={},
|
|
logging=LoggingCapability(),
|
|
prompts=PromptsCapability(list_changed=False),
|
|
resources=ResourcesCapability(subscribe=True, list_changed=False),
|
|
tools=ToolsCapability(list_changed=False),
|
|
completions=CompletionsCapability(),
|
|
)
|
|
)
|
|
|
|
|
|
@requirement("lifecycle:initialize:capabilities:minimal")
|
|
async def test_initialize_minimal_server_advertises_no_capabilities(connect: Connect) -> None:
|
|
async with connect(Server("bare")) as client:
|
|
capabilities = client.server_capabilities
|
|
|
|
assert capabilities == snapshot(ServerCapabilities(experimental={}))
|
|
|
|
|
|
@requirement("lifecycle:initialize:client-info")
|
|
async def test_initialize_server_sees_client_info(connect: Connect) -> None:
|
|
async def list_tools(
|
|
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
|
|
) -> types.ListToolsResult:
|
|
return types.ListToolsResult(
|
|
tools=[types.Tool(name="whoami", description="Report the caller.", input_schema={"type": "object"})]
|
|
)
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "whoami"
|
|
assert ctx.session.client_params is not None
|
|
client_info = ctx.session.client_params.client_info
|
|
return CallToolResult(content=[TextContent(text=f"{client_info.name} {client_info.version}")])
|
|
|
|
server = Server("introspector", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
async with connect(server, client_info=Implementation(name="acme-agent", version="9.9.9")) as client:
|
|
result = await client.call_tool("whoami", {})
|
|
|
|
assert result == snapshot(CallToolResult(content=[TextContent(text="acme-agent 9.9.9")]))
|
|
|
|
|
|
@requirement("lifecycle:initialize:client-capabilities")
|
|
async def test_initialize_server_sees_client_capabilities(connect: Connect) -> None:
|
|
async def list_tools(
|
|
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
|
|
) -> types.ListToolsResult:
|
|
return types.ListToolsResult(
|
|
tools=[types.Tool(name="abilities", description="Report capabilities.", input_schema={"type": "object"})]
|
|
)
|
|
|
|
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
|
|
assert params.name == "abilities"
|
|
assert ctx.session.client_params is not None
|
|
capabilities = ctx.session.client_params.capabilities
|
|
declared = [
|
|
name
|
|
for name, value in (
|
|
("sampling", capabilities.sampling),
|
|
("elicitation", capabilities.elicitation),
|
|
)
|
|
if value is not None
|
|
]
|
|
if capabilities.roots is not None:
|
|
declared.append(f"roots(list_changed={capabilities.roots.list_changed})")
|
|
return CallToolResult(content=[TextContent(text=",".join(declared) or "none")])
|
|
|
|
async def list_roots(context: ClientRequestContext) -> types.ListRootsResult:
|
|
raise NotImplementedError
|
|
|
|
server = Server("introspector", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
async with connect(server) as client:
|
|
result = await client.call_tool("abilities", {})
|
|
assert result == snapshot(CallToolResult(content=[TextContent(text="none")]))
|
|
|
|
async with connect(server, list_roots_callback=list_roots) as client:
|
|
result = await client.call_tool("abilities", {})
|
|
assert result == snapshot(CallToolResult(content=[TextContent(text="roots(list_changed=True)")]))
|
|
|
|
|
|
@requirement("lifecycle:requests-before-initialized")
|
|
async def test_request_before_initialization_is_rejected() -> None:
|
|
"""The server's rejection reason never reaches the client; it surfaces as a generic invalid-params error."""
|
|
|
|
async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
|
|
"""Registered so the request is routed to a real handler; never reached."""
|
|
raise NotImplementedError
|
|
|
|
server = Server("strict", on_list_tools=list_tools)
|
|
|
|
async with (
|
|
InMemoryTransport(server) as (read_stream, write_stream),
|
|
ClientSession(read_stream, write_stream) as session,
|
|
):
|
|
with anyio.fail_after(5):
|
|
with pytest.raises(MCPError) as exc_info:
|
|
await session.send_request(ListToolsRequest(), ListToolsResult)
|
|
|
|
# Ping is explicitly permitted before initialization completes.
|
|
pong = await session.send_ping()
|
|
|
|
assert exc_info.value.error == snapshot(
|
|
ErrorData(code=INVALID_PARAMS, message="Invalid request parameters", data="")
|
|
)
|
|
assert pong == snapshot(EmptyResult())
|
|
|
|
|
|
@requirement("lifecycle:version:match")
|
|
@requirement("lifecycle:version:server-fallback-latest")
|
|
async def test_initialize_negotiates_protocol_version() -> None:
|
|
server = Server("negotiator")
|
|
|
|
def initialize_request(protocol_version: str) -> InitializeRequest:
|
|
return InitializeRequest(
|
|
params=InitializeRequestParams(
|
|
protocol_version=protocol_version,
|
|
capabilities=ClientCapabilities(),
|
|
client_info=Implementation(name="time-traveller", version="0.0.1"),
|
|
)
|
|
)
|
|
|
|
async with (
|
|
InMemoryTransport(server) as (read_stream, write_stream),
|
|
ClientSession(read_stream, write_stream) as session,
|
|
):
|
|
with anyio.fail_after(5):
|
|
result = await session.send_request(initialize_request("2025-03-26"), InitializeResult)
|
|
assert result.protocol_version == snapshot("2025-03-26")
|
|
|
|
async with (
|
|
InMemoryTransport(server) as (read_stream, write_stream),
|
|
ClientSession(read_stream, write_stream) as session,
|
|
):
|
|
with anyio.fail_after(5):
|
|
result = await session.send_request(initialize_request("1999-01-01"), InitializeResult)
|
|
assert result.protocol_version == snapshot("2025-11-25")
|
|
|
|
|
|
@requirement("lifecycle:version:reject-unsupported")
|
|
async def test_unsupported_server_protocol_version_fails_initialization() -> None:
|
|
"""No real Server answers initialize with an unsupported version, so the server side is scripted by hand."""
|
|
|
|
async def scripted_server(streams: MessageStream) -> None:
|
|
server_read, server_write = streams
|
|
message = await server_read.receive()
|
|
assert isinstance(message, SessionMessage)
|
|
request = message.message
|
|
assert isinstance(request, JSONRPCRequest)
|
|
assert request.method == "initialize"
|
|
result = InitializeResult(
|
|
protocol_version="1991-08-06",
|
|
capabilities=ServerCapabilities(),
|
|
server_info=Implementation(name="relic", version="0.0.1"),
|
|
)
|
|
await server_write.send(
|
|
SessionMessage(
|
|
JSONRPCResponse(
|
|
jsonrpc="2.0",
|
|
id=request.id,
|
|
# Serialized exactly as a real server serializes results onto the wire.
|
|
result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
|
|
)
|
|
)
|
|
)
|
|
|
|
async with (
|
|
create_client_server_memory_streams() as ((client_read, client_write), server_streams),
|
|
anyio.create_task_group() as tg,
|
|
ClientSession(client_read, client_write) as session,
|
|
):
|
|
tg.start_soon(scripted_server, server_streams)
|
|
with anyio.fail_after(5):
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
await session.initialize()
|
|
|
|
assert str(exc_info.value) == snapshot("Unsupported protocol version from the server: 1991-08-06")
|
|
|
|
|
|
@requirement("lifecycle:version:downgrade")
|
|
async def test_an_older_supported_protocol_version_from_the_server_is_accepted() -> None:
|
|
"""A real Server echoes the requested version or its own latest, so the server side is scripted by hand."""
|
|
|
|
async def scripted_server(streams: MessageStream) -> None:
|
|
server_read, server_write = streams
|
|
message = await server_read.receive()
|
|
assert isinstance(message, SessionMessage)
|
|
request = message.message
|
|
assert isinstance(request, JSONRPCRequest)
|
|
assert request.method == "initialize"
|
|
result = InitializeResult(
|
|
protocol_version="2025-06-18",
|
|
capabilities=ServerCapabilities(),
|
|
server_info=Implementation(name="conservative", version="0.0.1"),
|
|
)
|
|
await server_write.send(
|
|
SessionMessage(
|
|
JSONRPCResponse(
|
|
jsonrpc="2.0",
|
|
id=request.id,
|
|
# Serialized exactly as a real server serializes results onto the wire.
|
|
result=result.model_dump(by_alias=True, mode="json", exclude_none=True),
|
|
)
|
|
)
|
|
)
|
|
|
|
async with (
|
|
create_client_server_memory_streams() as ((client_read, client_write), server_streams),
|
|
anyio.create_task_group() as tg,
|
|
ClientSession(client_read, client_write) as session,
|
|
):
|
|
tg.start_soon(scripted_server, server_streams)
|
|
with anyio.fail_after(5):
|
|
initialize_result = await session.initialize()
|
|
|
|
assert initialize_result.protocol_version == snapshot("2025-06-18")
|