LangChain工具与MCP

LangChain Tools工具

介绍

LangChain 的 工具（Tools） 是让大语言模型（LLM）能够与外部世界交互的核心机制。工具是一些封装了可调用函数及其输入结构的组件，允许 LLM 在运行时决定是否调用某个工具、以及以什么参数调用。

工具的核心作用

• 扩展模型能力：LLM 本身只能“说”，不能“做”。通过工具，模型可以执行如查询数据库、调用 API、读写文件、进行计算等操作。
• 结构化交互：工具定义了清晰的输入输出接口（通常基于类型提示和 Pydantic 模型），确保 LLM 调用时符合预期格式。
• 支持智能代理（Agent）：在 LangGraph 或 LangChain Agent 架构中，工具是 Agent 决策后执行动作的基本单元。

tools工具工作原理

创建工具

工具构成

• name：工具名称（字符串）
• description：工具描述（供 LLM 理解何时使用该工具）
• args_schema：工具参数
• func：实际执行的函数（接受字符串输入，返回字符串输出）

方式1：基础定义（推荐方式）

• 类型提示（Type hints） 是必需的，用于自动生成工具的输入 Schema。
• 函数 docstring 默认作为工具描述，帮助 LLM 理解何时使用该工具。

from langchain.tools import tool

@tool
def search_database(query: str, limit: int = 10) -> str:
    """Search the customer database for records matching the query.

    Args:
        query: Search terms to look for
        limit: Maximum number of results to return
    """
    return f"Found {limit} results for '{query}'"

方式2：自定义工具名称与描述

@tool("web_search", description="Use this to search the internet for current events.")
def search(query: str) -> str:
    return f"Results: {query}"

方式3：使用 Pydantic 定义复杂参数

• 当工具需要多个结构化参数时，可使用 args_schema 指定 Pydantic 模型：

from pydantic import BaseModel, Field
from langchain_core.tools import tool

class WeatherInput(BaseModel):
    location: str = Field(description="City or coordinates")
    unit: str = Field(default="celsius", description="Temperature unit")

@tool(args_schema=WeatherInput)
def get_weather(location: str, unit: str = "celsius") -> str:
    """Get current weather in a given location."""
    temp = 22 if unit == "celsius" else 72
    return f"Weather in {location}: {temp}°{unit[0].upper()}"

调用工具

通过agent调用工具

# 创建langchain工具示例

from langchain.tools import tool
from langchain.agents import create_agent

from my_model import model


@tool
def get_weather(city: str) -> str:
    """
    Get the weather in a city

    Args:
        city: The city to get the weather for
    """
    return f"The weather in {city} is nice. The temperature is 20 degrees, and it feels like 22 degrees."


tools = [get_weather]

# 创建智能体，调用工具
agent = create_agent(
    model=model, tools=tools, system_prompt="You are a helpful assistant"
)

# response = agent.invoke(
#     {"messages": [{"role": "user", "content": "佛山今天天气怎样？"}]}
# )
# print(response)

# 流式响应：实时获取 Agent 的思考和行动过程
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "佛山今天天气怎样？"}]},
    stream_mode="values",
):
    latest = chunk["messages"][-1]
    if latest.content:
        print(f"Answer: {latest.content}")
    elif latest.tool_calls:
        print(f"Calling: {[t['name'] for t in latest.tool_calls]}")

访问上下文与状态

工具不仅能接收用户输入，还能访问运行时上下文、对话状态或持久化存储

访问 Agent 状态（如消息历史）：使用 InjectedState 注入当前图状态（对 LLM 不可见）

from typing_extensions import Annotated
from langchain.tools import InjectedState, tool

@tool
def summarize_conversation(state: Annotated[dict, InjectedState]) -> str:
    messages = state["messages"]
    human_count = sum(1 for m in messages if m.__class__.__name__ == "HumanMessage")
    return f"You've sent {human_count} messages."

更新状态（返回 Command）

from langgraph.types import Command

@tool
def clear_conversation() -> Command:
    """Clear all messages."""
    return Command(update={"messages": []})

访问运行时上下文（如用户 ID）：通过 get_runtime() 获取不可变上下文

from langgraph.runtime import get_runtime
from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: str

@tool
def get_account_info() -> str:
    runtime = get_runtime(UserContext)
    user_id = runtime.context.user_id
    # 查询数据库...
    return f"User: {user_id}"

持久化存储（Store）跨会话保存用户数据：

@tool
def save_user_info(user_id: str, info: dict) -> str:
    store = get_runtime().store
    store.put(("users",), user_id, info)
    return "Saved!"

MCP与工具

介绍

MCP（Model Context Protocol） 是一个由 OpenRouter、Fireworks AI 等公司推动的新兴标准，用于统一 LLM 与外部工具/上下文的交互方式。MCP 是一个**开放协议**，用于标准化应用程序如何向大语言模型（LLM）提供工具（tools）和上下文（context）。

• ✅ 解耦工具实现与 LLM 调用逻辑
• ✅ 跨框架/语言复用工具服务（如 Python 工具可被 JavaScript Agent 使用）
• ✅ 支持远程、本地、流式等多种部署模式
• ✅ 统一工具发现、调用、状态管理接口

关键特点（来自 mcp.dev 官网）：

• 基于 JSON-RPC 2.0 协议
• 支持 工具列表发现（list tools）、调用（call tool）、流式结果
• 服务端实现（MCP Server）可独立部署（如 mcp-server-filesystem, mcp-server-notion）
• 客户端（LLM 或 Agent）通过标准接口连接多个 MCP 服务

MCP 的关键概念

• MCP 不关心工具用什么语言写——只要符合协议，Agent 就能调用。

概念	说明
MCP Server	实现具体工具逻辑的服务端（如天气查询、数据库访问）
MCP Client	Agent 侧的客户端，用于发现和调用远程/本地工具
Transport	通信方式：stdio（本地子进程）、streamable-http（HTTP/SSE）
Tool Schema	自动暴露工具名称、参数、描述（符合 OpenAPI 风格）

Agents 与 MCP 的关系

MCP 是 Agents 的“工具供给层”。 LangChain Agents 负责“是否调用工具、如何推理”，而 MCP 负责“如何安全、统一地调用具体工具”。

维度	LangChain Agents	MCP
定位	应用框架中的智能体实现	跨框架的工具调用通信协议
作用层级	高层应用逻辑（ReAct 循环、状态管理）	底层 I/O 接口（如何调用工具）
依赖关系	Agents 可以使用 MCP 作为工具后端	MCP 为 Agents 提供标准化工具接入

LangChain 集成 MCP

安装

pip install -U langchain-mcp-adapters -i https://mirrors.aliyun.com/pypi/simple --trusted-host=mirrors.aliyun.com

创建多服务器 MCP 客户端

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient({
    "math": {
        "transport": "stdio",
        "command": "python",
        "args": ["/path/to/math_server.py"]
    },
    "weather": {
        "transport": "streamable_http",
        "url": "http://localhost:8000/mcp"
    }
})

获取 MCP 工具并注入 Agent

from langchain.agents import create_agent

# 从 MCP 服务器加载工具
tools = await client.get_tools()  # 返回 List[BaseTool]

# 创建支持工具调用的 Agent
agent = create_agent(
    model="openai:gpt-4o",
    tools=tools
)

# 调用示例
response = await agent.ainvoke({
    "messages": [{"role": "user", "content": "What's (3 + 5) × 12?"}]
})

MCP server

创建MCP服务端：参考资料

"""
FastMCP quickstart example.

Run from the repository root:
    uv run examples/snippets/servers/fastmcp_quickstart.py
"""

from mcp.server.fastmcp import FastMCP

# Create an MCP server
mcp = FastMCP("Demo", json_response=True)


# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b


# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"


# Add a prompt
@mcp.prompt()
def greet_user(name: str, style: str = "friendly") -> str:
    """Generate a greeting prompt"""
    styles = {
        "friendly": "Please write a warm, friendly greeting",
        "formal": "Please write a formal, professional greeting",
        "casual": "Please write a casual, relaxed greeting",
    }

    return f"{styles.get(style, styles['friendly'])} for someone named {name}."


# Run with streamable HTTP transport
if __name__ == "__main__":
    mcp.run(transport="streamable-http")