mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-10-09 14:33:21 +02:00
3ca425416elangchain 1.4.2 ships `langchain.mcp`, whose MCPAdapter wraps fastmcp.Client and ClientGroup directly (resolve_tool, list_tools cache_mode, call_tool, Client.new, elicitation callbacks, MCPConfig, legacy and auto modes). The workflow now runs langchain's own MCP tests against the built wheels, and a new smoke drives MCPAdapter in-process, over stdio, MCPConfig, HTTP, SSE, a proxy, and a ClientGroup of a legacy HTTP server and a modern stdio server. Every smoke also runs through a FastMCP proxy in front of a stdio server and covers shapes this release cycle got wrong: `|` and non-ASCII template literals, comma-joined and exploded list query params, a float schema count, and a timed-out call followed by another on the same connection. Gaps the proxy already had in 4.0.5 are reported as known gaps that fail the run once they start passing. Server stderr goes to a log file, and the SDK's SSE reader is quieted where a check cancels a call on purpose. langchain-mcp-adapters' smoke is renamed smoke_langchain_mcp_adapters.py. Claude-Session: https://claude.ai/code/session_01KfHgVhbYEhBCC5eSeqGiuG Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
373 lines
13 KiB
Python
373 lines
13 KiB
Python
"""Drive FastMCP through `langchain.mcp`, LangChain's own MCP integration.
|
|
|
|
`langchain[mcp]` depends on `fastmcp` directly: `MCPAdapter` wraps a
|
|
`fastmcp.Client` or a `fastmcp.client.group.ClientGroup` and turns their tools
|
|
into LangChain tools. This runs in its own environment with the build under
|
|
test, against servers started from the checkout:
|
|
|
|
uv build --wheel --out-dir dist . fastmcp_slim
|
|
uv run --isolated --no-project --no-config --with 'langchain[mcp]' \\
|
|
--with "fastmcp @ file://$PWD/$(ls dist/fastmcp-4*.whl)" \\
|
|
--with "fastmcp-slim @ file://$PWD/$(ls dist/fastmcp_slim-*.whl)" \\
|
|
python tests/downstream/smoke_langchain_mcp.py
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import warnings
|
|
from collections.abc import Awaitable, Callable
|
|
from importlib.metadata import version
|
|
from typing import Any
|
|
|
|
import anyio
|
|
from _harness import (
|
|
LEGACY_PROXY_GAPS,
|
|
PROXY_GAPS,
|
|
TOOLS,
|
|
Report,
|
|
quiet,
|
|
serve,
|
|
server_env,
|
|
stdio_command,
|
|
stdio_transport,
|
|
)
|
|
from langchain.agents import create_agent
|
|
from langchain_core._api import LangChainBetaWarning
|
|
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
|
|
from langchain_core.messages import AIMessage, ToolMessage
|
|
|
|
import fastmcp
|
|
from fastmcp.client import Client
|
|
from fastmcp.client.group import ClientGroup
|
|
from fastmcp.mcp_config import MCPConfig
|
|
|
|
warnings.filterwarnings("ignore", category=LangChainBetaWarning)
|
|
from langchain.mcp import MCPAdapter # noqa: E402
|
|
|
|
MODERN = "2026-07-28"
|
|
|
|
|
|
class ScriptedToolModel(GenericFakeChatModel):
|
|
def bind_tools(self, tools: Any, **kwargs: Any) -> "ScriptedToolModel":
|
|
return self
|
|
|
|
|
|
def scripted_model(add: str, forecast: str) -> ScriptedToolModel:
|
|
return ScriptedToolModel(
|
|
messages=iter(
|
|
[
|
|
AIMessage(
|
|
content="",
|
|
tool_calls=[
|
|
{"name": add, "args": {"a": 2, "b": 3}, "id": "call-add"},
|
|
{
|
|
"name": forecast,
|
|
"args": {"city": "Chicago", "days": 2},
|
|
"id": "call-forecast",
|
|
},
|
|
],
|
|
),
|
|
AIMessage(content="done"),
|
|
]
|
|
)
|
|
)
|
|
|
|
|
|
async def call(tool: Any, args: dict[str, Any]) -> ToolMessage:
|
|
"""Invoke a tool the way an agent does, so the result is a ToolMessage with its artifact."""
|
|
return await tool.ainvoke(
|
|
{"type": "tool_call", "name": tool.name, "args": args, "id": "call-1"}
|
|
)
|
|
|
|
|
|
async def accept(message: str, response_type: Any, params: Any, context: Any) -> Any:
|
|
return {"approved": True}
|
|
|
|
|
|
class Observed:
|
|
"""Progress and log notifications a client received."""
|
|
|
|
def __init__(self) -> None:
|
|
self.progress: list[float] = []
|
|
self.logs: list[str] = []
|
|
|
|
async def on_progress(
|
|
self, value: float, total: float | None, message: str | None
|
|
) -> None:
|
|
self.progress.append(value)
|
|
|
|
async def on_log(self, message: Any) -> None:
|
|
self.logs.append(str(message.data))
|
|
|
|
def client(self, target: Any, **kwargs: Any) -> Client[Any]:
|
|
return Client(
|
|
target,
|
|
elicitation_handler=accept,
|
|
progress_handler=self.on_progress,
|
|
log_handler=self.on_log,
|
|
**kwargs,
|
|
)
|
|
|
|
|
|
async def run_checks(
|
|
report: Report,
|
|
checks: list[Callable[[], Awaitable[object]]],
|
|
gaps: dict[str, str] | None = None,
|
|
) -> None:
|
|
for check in checks:
|
|
name = check.__name__.replace("_", " ")
|
|
await report.check(name, check, known_gap=(gaps or {}).get(name))
|
|
|
|
|
|
async def exercise(
|
|
report: Report,
|
|
target: Any,
|
|
gaps: dict[str, str] | None = None,
|
|
**client_kwargs: Any,
|
|
) -> None:
|
|
seen = Observed()
|
|
async with MCPAdapter(seen.client(target, **client_kwargs)) as adapter:
|
|
tools: dict[str, Any] = {}
|
|
|
|
async def lists_tools() -> None:
|
|
tools.update({t.name: t for t in await adapter.list_tools()})
|
|
assert TOOLS <= set(tools), sorted(tools)
|
|
|
|
async def refreshes_tool_list() -> None:
|
|
refreshed = {t.name for t in await adapter.list_tools(cache_mode="refresh")}
|
|
assert TOOLS <= refreshed, sorted(refreshed)
|
|
|
|
async def server_metadata() -> None:
|
|
server = tools["add"].metadata["mcp"]["server"]
|
|
assert server["name"] == "downstream-smoke", server
|
|
|
|
async def agent_loop() -> None:
|
|
agent = create_agent(
|
|
scripted_model("add", "forecast"), list(tools.values())
|
|
)
|
|
state = await agent.ainvoke({"messages": [("user", "add, then forecast")]})
|
|
returned = {
|
|
m.name: m.content for m in state["messages"] if m.type == "tool"
|
|
}
|
|
assert "5" in str(returned["add"]), returned
|
|
assert "sunny" in str(returned["forecast"]), returned
|
|
|
|
async def structured_output() -> None:
|
|
message = await call(tools["forecast"], {"city": "Oslo"})
|
|
expected = {"city": "Oslo", "celsius": 21.5, "conditions": ["sunny"]}
|
|
assert message.artifact == {"structured_content": expected}, (
|
|
message.artifact
|
|
)
|
|
|
|
async def float_count_output_schema() -> None:
|
|
message = await call(tools["stamp"], {})
|
|
expected = {"when": "2026-09-22T00:00:00Z"}
|
|
assert message.artifact == {"structured_content": expected}, message
|
|
|
|
async def tool_error() -> None:
|
|
message = await call(tools["divide"], {"a": 1, "b": 0})
|
|
assert message.status == "error", message
|
|
assert "divide by zero" in str(message.content), message
|
|
|
|
async def image() -> None:
|
|
[block] = (await call(tools["snapshot"], {})).content
|
|
assert block["type"] == "image" and block["mime_type"] == "image/png", block
|
|
|
|
async def mixed_content() -> None:
|
|
blocks = (await call(tools["attachments"], {})).content
|
|
shapes = [(b["type"], b.get("text") or b.get("url")) for b in blocks]
|
|
assert shapes == [
|
|
("text", "see attached"),
|
|
("text", "buy milk"),
|
|
("file", "config://app"),
|
|
], shapes
|
|
|
|
async def elicitation() -> None:
|
|
message = await call(tools["confirm"], {"action": "deploy"})
|
|
assert "deploy: approved" in str(message.content), message
|
|
|
|
async def progress_notifications() -> None:
|
|
seen.progress.clear()
|
|
await call(tools["count_to"], {"n": 3})
|
|
assert seen.progress == [1, 2, 3], seen.progress
|
|
|
|
async def logging() -> None:
|
|
seen.logs.clear()
|
|
await call(tools["count_to"], {"n": 1})
|
|
assert any("counted to 1" in line for line in seen.logs), seen.logs
|
|
|
|
async def timed_out_call_leaves_connection_usable() -> None:
|
|
try:
|
|
with anyio.fail_after(0.5):
|
|
await call(tools["sleep"], {"seconds": 5})
|
|
except TimeoutError:
|
|
pass
|
|
else:
|
|
raise AssertionError("sleep(5) finished inside a 0.5s deadline")
|
|
with anyio.fail_after(10):
|
|
assert "2" in str((await call(tools["add"], {"a": 1, "b": 1})).content)
|
|
await anyio.sleep(0.2)
|
|
|
|
await run_checks(
|
|
report,
|
|
[
|
|
lists_tools,
|
|
refreshes_tool_list,
|
|
server_metadata,
|
|
agent_loop,
|
|
structured_output,
|
|
float_count_output_schema,
|
|
tool_error,
|
|
image,
|
|
mixed_content,
|
|
elicitation,
|
|
progress_notifications,
|
|
logging,
|
|
timed_out_call_leaves_connection_usable,
|
|
],
|
|
gaps,
|
|
)
|
|
|
|
|
|
async def exercise_group(report: Report, legacy_url: str, token: str) -> None:
|
|
"""One adapter over a ClientGroup: a handshake-era HTTP server and a modern stdio one."""
|
|
command, args = stdio_command()
|
|
legacy, modern = Observed(), Observed()
|
|
group = ClientGroup(
|
|
{
|
|
"legacy": legacy.client(legacy_url, auth=token, mode="legacy"),
|
|
"modern": modern.client(stdio_transport(), mode="auto"),
|
|
}
|
|
)
|
|
async with MCPAdapter(group) as adapter:
|
|
tools: dict[str, Any] = {}
|
|
|
|
async def lists_prefixed_tools() -> None:
|
|
tools.update({t.name: t for t in await adapter.list_tools()})
|
|
expected = {
|
|
f"{server}_{name}" for server in ("legacy", "modern") for name in TOOLS
|
|
}
|
|
assert expected <= set(tools), sorted(tools)
|
|
|
|
async def each_member_keeps_its_protocol_era() -> None:
|
|
members = adapter.client.clients # ty: ignore[unresolved-attribute]
|
|
assert members["legacy"].protocol_version not in (None, MODERN), members
|
|
assert members["modern"].protocol_version == MODERN, members
|
|
|
|
async def agent_loop_across_servers() -> None:
|
|
model = scripted_model("legacy_add", "modern_forecast")
|
|
agent = create_agent(model, list(tools.values()))
|
|
state = await agent.ainvoke({"messages": [("user", "add, then forecast")]})
|
|
returned = {
|
|
m.name: m.content for m in state["messages"] if m.type == "tool"
|
|
}
|
|
assert "5" in str(returned["legacy_add"]), returned
|
|
assert "sunny" in str(returned["modern_forecast"]), returned
|
|
|
|
async def elicitation_in_both_eras() -> None:
|
|
for server in ("legacy", "modern"):
|
|
message = await call(tools[f"{server}_confirm"], {"action": server})
|
|
assert f"{server}: approved" in str(message.content), (server, message)
|
|
|
|
async def calls_route_to_their_own_server() -> None:
|
|
legacy.progress.clear()
|
|
modern.progress.clear()
|
|
await call(tools["modern_count_to"], {"n": 2})
|
|
assert modern.progress == [1, 2] and legacy.progress == [], (
|
|
legacy.progress,
|
|
modern.progress,
|
|
)
|
|
|
|
async def tool_error_from_one_member() -> None:
|
|
message = await call(tools["legacy_divide"], {"a": 1, "b": 0})
|
|
assert message.status == "error", message
|
|
|
|
await run_checks(
|
|
report,
|
|
[
|
|
lists_prefixed_tools,
|
|
each_member_keeps_its_protocol_era,
|
|
agent_loop_across_servers,
|
|
elicitation_in_both_eras,
|
|
calls_route_to_their_own_server,
|
|
tool_error_from_one_member,
|
|
],
|
|
)
|
|
|
|
|
|
async def main() -> None:
|
|
report = Report(
|
|
"langchain.mcp",
|
|
{
|
|
"fastmcp": fastmcp.__version__,
|
|
"langchain": version("langchain"),
|
|
"mcp": version("mcp"),
|
|
},
|
|
)
|
|
command, args = stdio_command()
|
|
|
|
async def installed_build_is_under_test() -> None:
|
|
expected = os.environ.get("FASTMCP_EXPECTED_VERSION")
|
|
assert expected in (None, fastmcp.__version__), (
|
|
f"expected {expected}, got {fastmcp.__version__}"
|
|
)
|
|
|
|
with report.transport("in-process FastMCP server"):
|
|
from server import build
|
|
|
|
await report.check(
|
|
"installed build is under test", installed_build_is_under_test
|
|
)
|
|
await exercise(report, build())
|
|
|
|
with report.transport("stdio"):
|
|
await exercise(report, stdio_transport())
|
|
|
|
with report.transport("MCPConfig"):
|
|
|
|
async def adapter_from_config() -> None:
|
|
config = MCPConfig.from_dict(
|
|
{
|
|
"mcpServers": {
|
|
"smoke": {"command": command, "args": args, "env": server_env()}
|
|
}
|
|
}
|
|
)
|
|
async with MCPAdapter(config) as adapter:
|
|
names = {t.name for t in await adapter.list_tools()}
|
|
assert any(name.endswith("add") for name in names), sorted(names)
|
|
|
|
await report.check("adapter from MCPConfig", adapter_from_config)
|
|
|
|
for transport, label, client_kwargs, gaps in (
|
|
("http", "streamable HTTP", {}, None),
|
|
("sse", "SSE", {}, None),
|
|
("proxy", "FastMCP proxy", {}, PROXY_GAPS),
|
|
(
|
|
"proxy",
|
|
"FastMCP proxy, handshake-era client",
|
|
{"mode": "legacy"},
|
|
LEGACY_PROXY_GAPS,
|
|
),
|
|
):
|
|
# The SDK's SSE reader logs a traceback when a check cancels a call on purpose.
|
|
with (
|
|
report.transport(label),
|
|
serve(transport) as server,
|
|
quiet("mcp.client.sse"),
|
|
):
|
|
await exercise(
|
|
report, server.url, gaps=gaps, auth=server.token, **client_kwargs
|
|
)
|
|
|
|
with (
|
|
report.transport("ClientGroup: legacy HTTP + modern stdio"),
|
|
serve("http") as server,
|
|
):
|
|
await exercise_group(report, server.url, server.token)
|
|
|
|
report.finish()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|