MCP

什么是 MCP？

MCP（Model Context Protocol）是一种标准化协议，旨在简化大型语言模型（LLM）与外部数据源（如数据库、API 等）之间的交互。 通过 MCP，开发者可以让 LLM 以统一的方式访问和操作各种数据源，从而提升模型的实用性和扩展性。

MCP 的核心组件

1. MCP Server：负责处理来自 LLM 的请求，并将其转发到相应的数据源。它充当 LLM 和外部系统之间的中介。
2. MCP Client：集成在 LLM 中，用于发送请求到 MCP Server 并接收响应。
3. 数据源适配器：用于连接不同类型的数据源（如 SQL 数据库、NoSQL 数据库、RESTful API 等），并将数据转换为 MCP 可理解的格式。

MCP 的工作原理

1. 请求发送：LLM 通过 MCP Client 发送请求到 MCP Server，指定所需的数据操作（如查询、插入、更新等）。
2. 请求处理：MCP Server 接收到请求后，解析请求内容，并调用相应的数据源适配器。
3. 数据操作：数据源适配器与外部数据源进行交互，执行所需的操作。
4. 响应返回：操作完成后，数据源适配器将结果返回给 MCP Server，后者再将结果发送回 LLM。

MCP 的优势

• 标准化接口：通过统一的协议，简化了 LLM 与多种数据源的集成过程。
• 扩展性强：支持多种数据源类型，便于未来扩展和集成新的数据源。
• 提高效率：减少了开发者在集成数据源时的重复工作，提高了开发效率。
• 增强模型能力：使 LLM 能够访问更多实时和结构化的数据，提升其回答的准确性和相关性。

MCP 的应用场景

• 企业知识库：让 LLM 访问企业内部数据库，提供更准确的业务咨询和支持。
• 动态数据查询：通过 MCP 连接实时数据源，如股票市场数据、

天气信息等，增强模型的实时响应能力。

• 多数据源整合：在一个应用中集成多个数据源，如 CRM 系统、ERP 系统等，提供统一的查询接口。
• 定制化应用：根据特定业务需求，定制化开发 MCP 适配器，实现特定功能的数据交互。

---

实践示例

MCP Server 简单实现

下面是一个简单的 MCP Server 实现示例，使用 FastAPI 框架：

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Simple-Math-Server")


@mcp.tool()
def add_numbers(a: int, b: int) -> dict:
    """Add two numbers together and return a structured object."""
    total = a + b

    # Return a dictionary that matches your desired "properties" structure
    return {
        "result": total
    }


if __name__ == "__main__":
    mcp.run()

使用claude desktop app调用MCP Server

claude mcp 配置

{
  "mcpServers": {
    "my-math-tool": {
      "command": "python",
      "args": ["D:/project/PycharmProjects/fastApiProject/mcp-server.py"]
    }
  }
}

File Location:

Mac: ~/Library/Application Support/Claude/claude_desktop_config.json

Windows: %APPDATA%\Claude\claude_desktop_config.json

省略...

使用MCP Inspector来调用MCP Server

1. 启动MCP Server：

npx @modelcontextprotocol/inspector python mcp-server.py 

提示

pip install mcp 应该安装在全局环境，不能只在虚拟环境中安装，否则 Claude 找不到 MCP 包。

2. 在网页中连接

3. 调用 add_numbers 方法

使用langchain调用MCP Server

mcp-client.py

import asyncio
import sys
import operator
from typing import Annotated, TypedDict, List

# MCP SDK
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# LangChain / LangGraph
from langchain_ollama import ChatOllama
from langchain_core.tools import StructuredTool
from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage
from langgraph.graph import StateGraph, END

# --- 1. 定义状态 ---
class AgentState(TypedDict):
    messages: Annotated[List[BaseMessage], operator.add]

# --- 2. 转换器：把 MCP 工具变成 LangChain 工具 ---
def mcp_to_langchain(mcp_tool, session):
    async def _tool_func(**kwargs):
        # 调用 MCP Server
        result = await session.call_tool(mcp_tool.name, arguments=kwargs)
        # 提取结果文本
        if result.content and result.content[0].type == "text":
            return result.content[0].text
        return str(result)

    return StructuredTool.from_function(
        func=None,
        coroutine=_tool_func, # LangChain 支持异步工具
        name=mcp_tool.name,
        description=mcp_tool.description
    )

# --- 3. 主程序 ---
async def main():
    # 配置服务器启动参数 (确保 server.py 和 client.py 在同一目录)
    server_params = StdioServerParameters(
        command=sys.executable, 
        args=["D:/project/PycharmProjects/fastApiProject/mcp-server.py"], 
        env=None
    )

    print("🔌 Client: 正在连接 MCP Server...")

    # 建立连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化
            await session.initialize()
            
            # 获取工具
            tools_data = await session.list_tools()
            mcp_tools = tools_data.tools
            print(f"🛠️  Client: 发现工具 -> {[t.name for t in mcp_tools]}")

            # 转换工具给 LangChain
            lc_tools = [mcp_to_langchain(t, session) for t in mcp_tools]

            # 初始化 LLM (确保你的 Ollama 已经 pull 了 llama3.1 或 qwen2.5)
            llm = ChatOllama(model="llama3.1", temperature=0)
            llm_with_tools = llm.bind_tools(lc_tools)

            # --- 构建 LangGraph ---

            async def call_model(state: AgentState):
                # 调用 LLM
                response = await llm_with_tools.ainvoke(state["messages"])
                return {"messages": [response]}

            async def call_tools(state: AgentState):
                last_message = state["messages"][-1]
                results = []
                for call in last_message.tool_calls:
                    print(f"🤖 Agent: 决定调用工具 '{call['name']}' 参数: {call['args']}")
                    
                    # 查找并执行工具
                    tool = next((t for t in lc_tools if t.name == call['name']), None)
                    if tool:
                        output = await tool.coroutine(**call['args'])
                        print(f"✅ Agent: 工具返回结果 -> {output}")
                        
                        results.append(ToolMessage(
                            content=output,
                            tool_call_id=call["id"],
                            name=call["name"]
                        ))
                return {"messages": results}

            # 定义图结构
            workflow = StateGraph(AgentState)
            workflow.add_node("llm", call_model)
            workflow.add_node("tools", call_tools)
            workflow.set_entry_point("llm")

            # 条件边：有 tool_calls 就去 tools，否则结束
            workflow.add_conditional_edges(
                "llm",
                lambda s: "tools" if s["messages"][-1].tool_calls else END
            )
            workflow.add_edge("tools", "llm")

            agent = workflow.compile()

            # --- 运行测试 ---
            query = "请计算 100 加上 55 等于多少？"
            print(f"\n👤 用户: {query}")
            print("-" * 50)

            inputs = {"messages": [HumanMessage(content=query)]}
            
            # 运行图
            async for chunk in agent.astream(inputs, stream_mode="values"):
                # 只打印每一步最后一条消息的内容
                msg = chunk["messages"][-1]
                # print(f"[{msg.type}]: {msg.content}") 

            # 打印最终回复
            print("-" * 50)
            print(f"💡 最终答案: {chunk['messages'][-1].content}")

if __name__ == "__main__":
    asyncio.run(main())
    

输出：

🔌 Client: 正在连接 MCP Server...
🛠️  Client: 发现工具 -> ['add_numbers']

👤 用户: 请计算 100 加上 55 等于多少？
--------------------------------------------------
🤖 Agent: 决定调用工具 'add_numbers' 参数: {'a': 100, 'b': 55}
✅ Agent: 工具返回结果 -> {
  "result": 155
}
--------------------------------------------------
💡 最终答案: 100 加上 55 等于 155。