LangChain 高级用法之 MCP

1uv add langchain-mcp-adapters
2uv add fastmcp   # develop MCP server

 1import sys
 2
 3from fastmcp import FastMCP
 4
 5mcp = FastMCP("Math")
 6
 7@mcp.tool()
 8def add(a: int, b: int) -> int:
 9    """Add two numbers"""
10    print(f"math:add called with {a=} and {b=}", file=sys.stderr)
11    return a + b
12
13@mcp.tool()
14def multiply(a: int, b: int) -> int:
15    """Multiply two numbers"""
16    print(f"math:multiply called with {a=} and {b=}", file=sys.stderr)
17    return a * b
18
19if __name__ == "__main__":
20    mcp.run(transport="stdio", show_banner=False)

 1from fastmcp import FastMCP
 2
 3mcp = FastMCP("Weather")
 4
 5@mcp.tool()
 6async def get_weather(location: str) -> str:
 7    """Get weather for location."""
 8    print(f"Weather:get_weather called with {location=}")
 9    return "It's always sunny in New York"
10
11if __name__ == "__main__":
12    mcp.run(transport="streamable-http", show_banner=False)

 1import asyncio
 2from langchain_mcp_adapters.client import MultiServerMCPClient
 3from langchain.agents import create_agent
 4from pathlib import Path
 5
 6
 7async def main():
 8    math_server_path = str((Path(__file__).resolve().parent / "math_server.py"))
 9
10    client = MultiServerMCPClient(
11        {
12            "math": {
13                "transport": "stdio",  # Local subprocess communication
14                "command": "python",
15                # Absolute path to your math_server.py file
16                "args": [math_server_path],
17            },
18            "weather": {
19                "transport": "http",  # HTTP-based remote server
20                # Ensure you start your weather server on port 8000
21                "url": "http://localhost:8000/mcp",
22            }
23        }
24    )
25
26    tools = await client.get_tools()
27    agent = create_agent(
28        "ollama:gemma4:e4b",
29        tools=tools,
30    )
31    math_response = await agent.ainvoke(
32        {"messages": [{"role": "user", "content": "what's (3 + 5) x 12?"}]}
33    )
34
35    for message in math_response["messages"]:
36        message.pretty_print()
37
38    weather_response = await agent.ainvoke(
39        {"messages": [{"role": "user", "content": "what is the weather in nyc?"}]}
40    )
41
42    for message in weather_response["messages"]:
43        message.pretty_print()
44
45if __name__ == "__main__":
46    asyncio.run(main())

1uv run python src/langchain_study/mcp/weather_server.py
2[04/22/26 16:02:46] INFO     Starting MCP server 'Weather' with transport 'streamable-http' on http://127.0.0.1:8000/mcp                                                                                                            transport.py:301
3INFO:     Started server process [99625]
4INFO:     Waiting for application startup.
5INFO:     Application startup complete.
6INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

 1uv run python src/langchain_study/mcp/mcp_client_agent.py 
 2[04/22/26 16:06:20] INFO     Starting MCP server 'Math' with transport 'stdio'                                                                                                                                                      transport.py:209
 3[04/22/26 16:06:39] INFO     Starting MCP server 'Math' with transport 'stdio'                                                                                                                                                      transport.py:209
 4[04/22/26 16:06:39] INFO     Starting MCP server 'Math' with transport 'stdio'                                                                                                                                                      transport.py:209
 5math:multiply called with a=8 and b=12
 6math:add called with a=3 and b=5
 7================================ Human Message =================================
 8
 9what's (3 + 5) x 12?
10================================== Ai Message ==================================
11Tool Calls:
12  add (497dc421-b912-4664-82a4-e7ca8cfbc5e2)
13 Call ID: 497dc421-b912-4664-82a4-e7ca8cfbc5e2
14  Args:
15    a: 3
16    b: 5
17================================= Tool Message =================================
18Name: add
19
20[{'type': 'text', 'text': '8', 'id': 'lc_d3ef66f9-d7ba-4aeb-86fb-2548bd43d9bf'}]
21================================== Ai Message ==================================
22Tool Calls:
23  multiply (f242adcd-d1a1-4687-be75-a0a307ca32fe)
24 Call ID: f242adcd-d1a1-4687-be75-a0a307ca32fe
25  Args:
26    a: 8
27    b: 12
28================================= Tool Message =================================
29Name: multiply
30
31[{'type': 'text', 'text': '96', 'id': 'lc_dc22f392-4fb2-45a4-a57a-e00a5de2d960'}]
32================================== Ai Message ==================================
33
3496
35================================ Human Message =================================
36
37what is the weather in nyc?
38================================== Ai Message ==================================
39Tool Calls:
40  get_weather (cc8dd3bd-cb54-4856-808a-1f3979fda25c)
41 Call ID: cc8dd3bd-cb54-4856-808a-1f3979fda25c
42  Args:
43    location: nyc
44================================= Tool Message =================================
45Name: get_weather
46
47[{'type': 'text', 'text': "It's always sunny in New York", 'id': 'lc_aa46f91e-37a9-48bf-8959-558d233a44fe'}]
48================================== Ai Message ==================================
49Tool Calls:
50  get_weather (6d8a8492-22eb-4a72-b8f2-3f82194276e6)
51 Call ID: 6d8a8492-22eb-4a72-b8f2-3f82194276e6
52  Args:
53    location: nyc
54================================= Tool Message =================================
55Name: get_weather
56
57[{'type': 'text', 'text': "It's always sunny in New York", 'id': 'lc_07da67af-1ce3-474e-9199-8b7be3a6769e'}]
58================================== Ai Message ==================================

1from mcp.client.stdio import stdio_client
2from mcp.client.sse import sse_client
3from mcp.client.websocket import websocket_client
4from mcp.client.streamable_http import streamable_http_client

1from langchain.tools import tool
2
3@tool
4def add(a: int, b: int) -> int:
5    """Add two numbers"""
6    return a + b

 1client = MultiServerMCPClient(
 2    {
 3        "weather": {
 4            "transport": "http",
 5            "url": "http://localhost:8000/mcp",
 6            "headers": {
 7                "Authorization": "Bearer YOUR_TOKEN",
 8                "X-Custom-Header": "custom-value"
 9            },
10        }
11    }
12)

1client = MultiServerMCPClient(
2    {
3        "weather": {
4            "transport": "http",
5            "url": "http://localhost:8000/mcp",
6            "auth": auth,
7        }
8    }
9)

 1import base64
 2
 3import httpx
 4from langchain.agents import create_agent
 5from langchain_core.messages import HumanMessage
 6from langchain_core.tools import tool
 7
 8
 9@tool
10def fetch_image(image_url: str) -> list[dict]:
11    """Fetch image from the web"""
12    response = httpx.get(image_url)
13    content_type = response.headers.get("content-type", "image/png")
14    b64_data = base64.b64encode(response.content).decode("utf-8")
15    return [{"type": "image_url", "image_url": {"url": f"data:{content_type};base64,{b64_data}"}}]
16
17
18agent = create_agent(
19    # "ollama:gemma4:e4b",
20    "bedrock:us.anthropic.claude-haiku-4-5-20251001-v1:0",
21    tools=[fetch_image],
22)
23
24result = agent.invoke({"messages": [HumanMessage(content=(
25    "describe the the image at https://yanbin.blog/my-first-langchain-ai-agent/my-first-ai-agent-cats.png"
26    ))]})
27
28print(result["messages"][-1].content)

 1This image shows a Telegram chat interface with a bot called "Seek Cat." Here's what it displays:
 2
 3**Top portion:**
 4- A Telegram conversation header showing "Seek Cat bot" with a blue circular avatar containing an "S"
 5- A high-quality close-up photo of a tabby cat's face with striking yellow/green eyes and prominent whiskers, timestamped 6:29 PM
 6
 7**Bottom portion:**
 8- A message bubble announcing "🐱 New Cat Found!" with details about a cat named **Ribeye**:
 9  - **Sex:** Male/Neutered
10  - **Breed:** Domestic Shorthair/Mix
11  - **Age:** 1 year 9 months
12  - Links to "View Photo" and "View Full Profile"
13  - Below that is another cat photo showing a tabby and white cat lying in a cat hammock/bed, looking at the camera, also timestamped 6:29 PM
14
15**Background:** The chat window has a light green background with cute cat-themed emoji patterns (cats, fish, watermelons, etc.)
16
17This appears to be a demo or example screenshot from a blog post about creating an AI agent using LangChain, showing how the bot can present information about cats in a Telegram interface.

 1{
 2  "artifact": null,
 3  "content": [{"type": "image_url","image_url":  {"url":"data:image/png;base64,iVBORw0KGgoAAAAN......"}}],
 4  "content_blocks": [{"mime_type": "image/png", "type": "image", "base64": "iVBORw0KGgoAAAAN......", 
 5    "id": "lc_32058fe8-d4b8-4aad-bdfa-e0b88473becb"}],
 6  "name": "fetch_image",
 7  "text": "",
 8  "tool_call_id": "toolu_bdrk_01VFAFDYiu8jjnEb4Vf6QmkW",
 9  "type": "tool"
10}

 1@mcp.tool()
 2def fetch_image(image_url: str) -> ImageContent:
 3    """Fetch image from the web"""
 4    print(f"Image:fetch_image called with {image_url=}", file=sys.stderr)
 5    response = httpx.get(image_url)
 6    content_type = response.headers.get("content-type", "image/png")
 7    b64_data = base64.b64encode(response.content).decode("utf-8")
 8
 9    return ImageContent(
10            type="image",
11            data=b64_data,
12            mimeType=content_type
13        )

1@mcp.resource("resource://users/{user_id}")
2def get_user(user_id: str) -> str:
3    ...

1@mcp.prompt()
2def translate(text: str, target_lang: str = "English") -> str:
3    ...

 1async def mcp_tool_interceptor(request: MCPToolCallRequest, handler):
 2    modified_request = request.override(
 3        args={**(request.args | { "a":100})}
 4    )
 5
 6    return await handler(modified_request)
 7
 8client = MultiServerMCPClient(
 9    {...},
10    tool_interceptors=[mcp_tool_interceptor],
11)

 1@tool
 2def where_is_bob(runtime:  ToolRuntime) -> str:
 3    """Tell Bob where he is."""
 4    # print(runtime.state["messages"])
 5
 6    if runtime.context.user_id == "123":
 7        return Command(update={
 8            "messages": [
 9                ToolMessage("he is in SF", tool_call_id=runtime.tool_call_id)
10            ]
11        })
12    return "he is in Chicago"

 1async def mcp_tool_interceptor(request: MCPToolCallRequest, handler):
 2    result = await handler(request)
 3
 4    if request.name == "submit_order":
 5        return Command(
 6            update={
 7                "messages": [result] if isinstance(result, ToolMessage) else [],
 8                "task_status": "completed",
 9            },
10            goto="summary_agent",  # goto="__end__" 可提前结束执行
11        )
12
13    return result

 1import asyncio
 2
 3from fastmcp import FastMCP, Context
 4
 5mcp = FastMCP("web_fetch")
 6
 7@mcp.tool()
 8async def fetch(ctx: Context, url: str) -> str:
 9    """fetch web string content by url"""
10    await ctx.report_progress(30, total=100, message=f"loading {url=}")
11    await asyncio.sleep(1)
12    await ctx.report_progress(65, total=100, message=f"loading {url=}")
13    await asyncio.sleep(1)
14    await ctx.report_progress(100, total=100, message=f"loaded {url=}")
15    return "I'm a cat"
16
17if __name__ == "__main__":
18    mcp.run(transport="stdio", show_banner=False)

 1import asyncio
 2
 3from fastmcp import Client
 4
 5
 6async def progress_handler(progress: float, total: float | None, message: str) -> None:
 7    print(f"progress: {progress}/{total} - {message}")
 8
 9async def main():
10    client = Client("web_fetch.py", progress_handler=progress_handler)
11
12    async with client:
13        result = await client.call_tool("fetch", {"url": "https://yanbin.blog"})
14        print(result)
15
16if __name__ == "__main__":
17    asyncio.run(main())

1if __name__ == "__main__":
2    mcp.run(transport="http", show_banner=False) # 用 'streamable-http' 也是一样的

1uv run python src/langchain_study/mcp/web_fetch.py
2[04/23/26 21:13:21] INFO     Starting MCP server 'web_fetch' with transport 'http' on http://127.0.0.1:8000/mcp 

1client = Client("http://localhost:8000/mcp", progress_handler=progress_handler)

1if __name__ == "__main__":
2    mcp.run(transport="sse", show_banner=False)

1uv run python src/langchain_study/mcp/web_fetch.py
2[04/23/26 21:16:23] INFO     Starting MCP server 'web_fetch' with transport 'sse' on http://127.0.0.1:8000/sse 

1client = Client("http://localhost:8000/sse", progress_handler=progress_handler)

1if __name__ == "__main__":
2    mcp.run(transport="stdio", show_banner=False)

 1import asyncio
 2
 3from langchain.agents import create_agent
 4from langchain_mcp_adapters.callbacks import CallbackContext, Callbacks
 5from langchain_mcp_adapters.client import MultiServerMCPClient
 6
 7
 8async def on_progress(progress: float, total: float | None, message: str | None, context: CallbackContext):
 9    print(f"{context.server_name}:{context.tool_name} - progress: {progress}/{total} - {message}")
10
11async def main():
12    client = MultiServerMCPClient(
13        {
14            "web_fetch": {
15                "transport": "stdio",
16                "command": "python",
17                "args": ["web_fetch.py"],
18            }
19        },
20        callbacks=Callbacks(on_progress=on_progress),
21    )
22
23    tools = await client.get_tools()
24    agent = create_agent(
25        "ollama:gemma4:e4b",
26        tools=tools,
27    )
28
29    await agent.ainvoke({"messages": [{"role": "user", "content": "read web content from https://yanbin.blog"}]})
30
31
32if __name__ == "__main__":
33    asyncio.run(main())

1    client = MultiServerMCPClient(
2        {
3            "web_fetch": {
4                "transport": "http",
5                "url": "http://localhost:8000/mcp"
6            }
7        },
8        callbacks=Callbacks(on_progress=on_progress),
9    )

1    client = MultiServerMCPClient(
2        {
3            "web_fetch": {
4                "transport": "sse",
5                "url": "http://localhost:8000/sse"
6            }
7        },
8        callbacks=Callbacks(on_progress=on_progress),
9    )

1uv run python src/langchain_study/mcp/langchain_mcp_client.py 
2web_fetch:fetch - progress: 30.0/100.0 - loading url='https://yanbin.blog'
3web_fetch:fetch - progress: 65.0/100.0 - loading url='https://yanbin.blog'
4web_fetch:fetch - progress: 100.0/100.0 - loaded url='https://yanbin.blog'

 1@mcp.tool()
 2async def fetch(ctx: Context, url: str) -> str:
 3    """fetch web string content by url"""
 4
 5    await ctx.debug(f"fetch debug log for {url=}", extra={"foo": "bar"})
 6    await ctx.info(f"fetch called with {url=}")
 7    await ctx.warning(f"fetch debug log for {url=}")
 8    await ctx.error(f"fetch debug log for {url=}")
 9
10    return "I'm a cat"

 1import asyncio
 2
 3from langchain.agents import create_agent
 4from langchain_mcp_adapters.callbacks import CallbackContext, Callbacks, LoggingMessageNotificationParams
 5from langchain_mcp_adapters.client import MultiServerMCPClient
 6
 7async def on_logging_message(params: LoggingMessageNotificationParams, context: CallbackContext):
 8    print(f"[{context.server_name}] {params.level}: {params.data}")
 9
10async def main():
11    client = MultiServerMCPClient(
12        {
13            "web_fetch": {
14                "transport": "sse",
15                "url": "http://localhost:8000/sse"
16            }
17        },
18        callbacks=Callbacks(on_logging_message=on_logging_message),
19    )
20
21    tools = await client.get_tools()
22    agent = create_agent("ollama:gemma4:e4b", tools=tools)
23
24    await agent.ainvoke({"messages": [{"role": "user", "content": "read web content from https://yanbin.blog"}]})
25
26
27if __name__ == "__main__":
28    asyncio.run(main())

1[web_fetch] debug: {'msg': "fetch debug log for url='https://yanbin.blog'", 'extra': {'foo': 'bar'}}
2[web_fetch] info: {'msg': "fetch called with url='https://yanbin.blog'", 'extra': None}
3[web_fetch] warning: {'msg': "fetch debug log for url='https://yanbin.blog'", 'extra': None}
4[web_fetch] error: {'msg': "fetch debug log for url='https://yanbin.blog'", 'extra': None}

 1from fastmcp import FastMCP, Context
 2from pydantic import BaseModel
 3
 4mcp = FastMCP("web_fetch")
 5
 6class QueryParams(BaseModel):
 7    id: int
 8    date: str
 9
10@mcp.tool()
11async def fetch(ctx: Context, url: str) -> str:
12    """fetch web string content by url"""
13
14    result = await ctx.elicit(message=f"Please provide id and date", response_type=QueryParams)
15    if result.action == "accept":
16        print(f"fetch called with {url=} and {result.data=}")
17    if result.action == "decline":
18        print(f"fetch called with {url=} but user declined to provide query params")
19    return f"I'm a cat: {result}"
20
21if __name__ == "__main__":
22    mcp.run(transport="sse", show_banner=False)

 1import asyncio
 2
 3from langchain.agents import create_agent
 4from langchain_mcp_adapters.callbacks import CallbackContext, Callbacks, LoggingMessageNotificationParams
 5from langchain_mcp_adapters.client import MultiServerMCPClient
 6from mcp.shared.context import RequestContext
 7from mcp.types import ElicitRequestParams, ElicitResult
 8
 9
10async def on_elicitation(mcp_context: RequestContext, params: ElicitRequestParams,
11        context: CallbackContext, ) -> ElicitResult:
12    return ElicitResult(
13        action="accept",
14        content={"id": 12345, "date": "2022-12-23"},
15    )
16
17async def main():
18    client = MultiServerMCPClient(
19        {
20            "web_fetch": {
21                "transport": "sse",
22                "url": "http://localhost:8000/sse"
23            }
24        },
25        callbacks=Callbacks(on_elicitation=on_elicitation),
26    )
27
28    tools = await client.get_tools()
29    agent = create_agent("ollama:gemma4:e4b", tools=tools)
30
31    result = await agent.ainvoke({"messages": [{"role": "user", "content": "read web content from https://yanbin.blog"}]})
32    print(result["messages"][-1].content)
33
34
35if __name__ == "__main__":
36    asyncio.run(main())

1uv run python src/langchain_study/mcp/langchain_mcp_client.py 
2I attempted to read the web content from `https://yanbin.blog`, but the data retrieved appears to be technical or incomplete, and does not contain the readable blog content.
3
4The content I received was:
5`I'm a cat: action='accept' data=QueryParams(id=12345, date='2022-12-23')`
6
7If you were looking for specific information, please let me know, and I can try a different approach.

1async def on_elicitation(mcp_context: RequestContext, params: ElicitRequestParams,
2        context: CallbackContext, ) -> ElicitResult: