1.MCP

MCP即Model Context Protocol（模型上下文协议），于 2024 年 11 月由 Claude 大模型的公司 Anthropic 推出并开源，提供统一的通信接口，旨在解决大模型与外部数据源、工具及系统集成时的碎片化问题，MCP 可以理解为AI 世界的 USB-C 接口，它的核心目标是让 AI 模型（如 ChatGPT、Claude 等）能够像通用遥控器一样，通过统一协议连接各类数据源和工具，实现高效、安全的跨系统协作。

MCP可以解决Function calling的弊端，Function calling在使用时需要构建一个外部函数作为中介，一边接收大模型发出的请求，另一边负责调用外部工具，通过这种方式使大模型能够间接实现对外部工具的调用。

比如当查询天气时，大模型调用外部工具的function calling过程如下：

此时需要编写外部函数，外部函数往往需要上百行代码，而且还需要设计提示词模版，才能提高Function calling响应的准确率。

而MCP能在Agent开发过程中，让大模型更加便捷的调用外部工具。MCP首先统一了Function calling的运行规范。MCP把大模型运行环境称作 MCP Client，也就是MCP客户端，同时，把外部函数运行环境称作MCP Server，也就是MCP服务器，然后统一了MCP客户端和服务器的运行规范，并且要求MCP客户端和服务器之间，也统一按照某个既定的提示词模板进行通信。

2.工作机制

以下案例给⼤模型提供路线规划的能⼒。
以⾼德地图提供的MCP服务为例，⾼德地图开放平台提供了MCP接⼊介绍： https://lbs.amap.com/api/mcp-server/summary。 只需要在⾼德开放平台注册账号，登录后，在左侧菜单中选择“应⽤管理”->“我的应⽤”，然后创建新应⽤，就可以申请api_key

from langchain_openai import ChatOpenAI
import datetime
from langchain.tools import tool

llm = ChatOpenAI(
    model="qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    openai_api_key="API_KEY",
    streaming=True  
)

# 定义工具
@tool(description="规划行车路线")
def get_route_plan(origin_city: str, target_city: str):
    """规划行车路线
    Args:
        origin_city: 出发城市
        target_city: 目标城市
    """
    result = f"从城市 {origin_city} 出发，到目标城市 {target_city} ,只需意念，三分钟即可到达。"
    print(">>>> get_route_plan >>>>>" + result)
    return result

# 大模型绑定工具
llm_with_tools = llm.bind_tools([get_route_plan])

# 工具容器
all_tools = {"get_route_plan": get_route_plan}

# 准备查询和消息列表
query = "帮我规划一条从广州到新疆的自驾路线"
messages = [query]

# 获取工具调用请求
ai_msg = llm_with_tools.invoke(messages)
print(ai_msg)
messages.append(ai_msg)

# 打印并处理工具调用
print(ai_msg.tool_calls)
if ai_msg.tool_calls:
    for tool_call in ai_msg.tool_calls:
        selected_tool = all_tools[tool_call["name"].lower()]
        tool_msg = selected_tool.invoke(tool_call)
        messages.append(tool_msg)

# 获取最终响应
print("\n流式输出结果：")
for chunk in llm_with_tools.stream(messages): 
    if chunk.content:  
        print(chunk.content, end="", flush=True)  

从结果可以看出高德提供的 MCP 服务数据可信度更高，所以被大模型直接采用它的数据，自己提供的意念传送相关数据缺乏可靠性，大模型参考了，但是最终没有采纳。可以知道工具在处理具体问题时存在的缺陷，是否调用工具完全由大模型自主决定，无论工具返回什么，最终大模型输出的内容还是自己说了算，至于工具在执行过程中的具体操作，大模型则不干预。

3.MCP的两种实现方式

分别是基于标准输入输出（stdio）的本地进程通信和基于 HTTP/SSE 的网络通信，以高德地图的 MCP 服务为例，其开放平台的说明中给出了两种接入该服务的配置方式。
sse方式

{
 "mcpServers": {
   "amap-amap-sse": {
     "url": "https://mcp.amap.com/sse?key=key"
   }
}
}

stdio方式

"amap-maps": {
      "command": "npx",
      "args": [
          "-y",
          "@amap/amap-maps-mcp-server"
      ],
      "env": {
          "AMAP_MAPS_API_KEY": "key"
      }
  }

两种方式区别

实际使用时，本地开发常用stdio方式，生产环境的分布式场景则更多使用HTTP/SSE方式。

4.Agent接入MCP服务

首先安装依赖

pip install langchain-mcp-adapters

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_community.chat_models import ChatTongyi
import asyncio

llm = ChatTongyi(
    model="qwen-plus",
    api_key="API_KEY",
)

async def main():
    # 配置高德地图MCP客户端，使用SSE方式连接
    client = MultiServerMCPClient(
        {
            "amap-maps-sse": {
                "url": "https://mcp.amap.com/sse?key=xxxxxxxx",
                "transport": "sse"
            },
             #  "amap-maps": {
               #     "command": "npx",
               #     "args": [
               #         "-y",
               #         "@amap/amap-maps-mcp-server"
               #     ],
               #     "env": {
               #         "AMAP_MAPS_API_KEY": "xxxxxxxx"
               #     },
               #     "transport":"stdio"
               # }
        }
    )
    try:
        tools = await client.get_tools()
        agent = create_react_agent(
            model=llm,
            tools=tools
        )
        response = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "帮我规划一条从广州到新疆的自驾路线"}]}
        )
        return response
    except Exception as e:
        print(f"发生错误: {e}")
        return None

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

5.手写MCP服务

MCP服务

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("roymcpdemo")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    print(f"roy mcp demo called : add({a}, {b})")
    return a + b


@mcp.tool()
def weather(city: str):
    """获取某个城市的天气
    Args:
        city: 具体城市
    """
    return "城市" + city + ",今天天气不错"


@mcp.resource("greeting://{name}")
def greeting(name: str) -> str:
    """Greet a person by name."""
    print(f"roy mcp demo called : greeting({name})")
    return f"Hello, {name}!"


if __name__ == "__main__":
    # 以sse协议暴露服务。
    mcp.run(transport='sse')

    # 以stdio协议暴露服务。
    # mcp.run(transport='stdio')

运行服务端

MCP客户端

from mcp import  ClientSession
import mcp.types as types

SERVER_URL = "http://127.0.0.1:8000/sse"

async def handle_sampling_message(message: types.CreateMessageRequestParams) -> types.CreateMessageResult:
    print(f"sampling message: {message}")
    return types.CreateMessageResult(
        role="assistant",
        content=types.TextContent(
            type="text",
            text="Hello,world! from model"
        ),
        model="qwen-plus",
        stopReason="endTurn"
    )


async def run():
    from mcp.client.sse import sse_client
    async with sse_client(SERVER_URL) as (read, write):
        async with ClientSession(read, write, sampling_callback=handle_sampling_message) as session:
            await session.initialize()

            prompts = await session.list_prompts()
            print(f"prompts: {prompts}")

            tools = await session.list_tools()
            print(f"tools: {tools}")

            resources = await session.list_resources()
            print(f"resources: {resources}")

            result = await session.call_tool("weather", {"city": "北京"})
            print(f"result: {result}")


if __name__ == "__main__":
    import asyncio

    asyncio.run(run())

调用MCP服务端