Max
50 lines
2.5 KiB
Python
50 lines
2.5 KiB
Python
"""Probe server/discover once, persist the result, reconnect with zero round-trips — a fresh `Client` via `targets`."""
|
|
|
|
from mcp.client import Client
|
|
from mcp.types import DiscoverResult
|
|
from mcp.types.version import LATEST_MODERN_VERSION
|
|
from stories._harness import TargetFactory, run_client
|
|
|
|
|
|
async def main(targets: TargetFactory, *, mode: str = "auto") -> None:
|
|
# The caller's mode (the real-user "auto" default) probes server/discover inside
|
|
# __aenter__ and caches the result; a hard version pin would skip the probe and
|
|
# never see the server's real DiscoverResult.
|
|
async with Client(targets(), mode=mode) as client:
|
|
discovered = client.session.discover_result
|
|
assert discovered is not None, "mode='auto' against a modern server populates discover_result"
|
|
assert client.protocol_version == LATEST_MODERN_VERSION
|
|
# On the 2026 era, server identity is an optional serverInfo stamp in the
|
|
# result _meta; an anonymous server reads as None. This one stamps it.
|
|
info = client.server_info
|
|
assert info is not None, "the server stamps serverInfo into its results"
|
|
assert info.name == "reconnect-example"
|
|
assert LATEST_MODERN_VERSION in discovered.supported_versions
|
|
|
|
result = await client.call_tool("add", {"a": 2, "b": 3})
|
|
assert result.structured_content == {"result": 5}, result
|
|
|
|
# Round-trip through JSON to model loading the result from an on-disk cache.
|
|
saved = discovered.model_dump_json(by_alias=True)
|
|
rehydrated = DiscoverResult.model_validate_json(saved)
|
|
assert rehydrated == discovered
|
|
|
|
# Reconnect: a version pin plus the cached DiscoverResult adopts the prior state with
|
|
# zero round-trips on entry. A Client cannot be re-entered after exit, so targets()
|
|
# yields a fresh one. Without prior_discover= a bare pin would leave server_info
|
|
# None — the cache is what carries the server's identity stamp across reconnects.
|
|
async with Client(targets(), mode=LATEST_MODERN_VERSION, prior_discover=rehydrated) as second:
|
|
assert second.protocol_version == LATEST_MODERN_VERSION
|
|
info = second.server_info
|
|
assert info is not None, "the cached DiscoverResult carries the serverInfo stamp"
|
|
assert info.name == "reconnect-example"
|
|
assert second.server_capabilities.tools is not None
|
|
assert second.session.discover_result == rehydrated
|
|
|
|
result = await second.call_tool("add", {"a": 1, "b": 1})
|
|
assert result.structured_content == {"result": 2}, result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_client(main)
|