MCP

MCP（Model Context Protocol，模型上下文协议） ，2024年11月底，由 Anthropic 推出的一种开放标准，旨在统一大模型与外部数据源和工具之间的通信协议。MCP 的主要目的在于解决当前 AI 模型因数据孤岛限制而无法充分发挥潜力的难题，MCP 使得 AI 应用能够安全地访问和操作本地及远程数据，为 AI 应用提供了连接万物的接口。

文档：Introduction - Model Context Protocol

MCP 可以直接在 AI 与数据（包括本地数据和互联网数据）之间架起一座桥梁，通过 MCP 服务器和 MCP 客户端，大家只要都遵循这套协议，就能实现“万物互联”。

MCP 与 Function Calling 的区别

● MCP（Model Context Protocol），模型上下文协议

● Function Calling，函数调用

这两种技术都旨在增强 AI 模型与外部数据的交互能力。

MCP 的核心技术

架构设计

MCP 遵循客户端-服务器架构（client-server），其中包含以下几个核心概念：

1.  MCP 主机（MCP Hosts）：发起请求的 LLM 应用程序（例如 Claude 、IDE 或 AI 工具）。

2.  MCP 客户端（MCP Clients）：在主机程序内部，与 MCP server 保持 1:1 的连接。

3.  MCP 服务器（MCP Servers）：为 MCP client 提供上下文、工具和 prompt 信息。

4.  本地资源（Local Resources）：本地计算机中可供 MCP server 安全访问的资源（例如文件、数据库）。

5.  远程资源（Remote Resources）：MCP server 可以连接到的远程资源（例如通过 API）。

MCP Client

MCP client 充当 LLM 和 MCP server 之间的桥梁，MCP client 的工作流程如下：

● MCP client 首先从 MCP server 获取可用的工具列表。

● 将用户的查询连同工具描述通过 function calling 一起发送给 LLM。

● LLM 决定是否需要使用工具以及使用哪些工具。

● 如果需要使用工具，MCP client 会通过 MCP server 执行相应的工具调用。

● 工具调用的结果会被发送回 LLM。

● LLM 基于所有信息生成自然语言响应。

● 最后将响应展示给用户。

MCP Server

MCP server 是 MCP 架构中的关键组件，它可以提供 3 种主要类型的功能：

● 资源（Resources）：类似文件的数据，可以被客户端读取，如 API 响应或文件内容。

● 工具（Tools）：可以被 LLM 调用的函数（需要用户批准）。

● 提示（Prompts）：预先编写的模板，帮助用户完成特定任务。

通信机制

两种主要的通信机制：基于标准输入输出的本地通信和基于SSE（Server-Sent-Events）的远程通信。这两种机制都使用 JSON-RPC 2.0格式进行消息传输，确保了通信的标准化和可扩展性。

● 本地通信：通过 stdio 传输数据，适用于在同一台机器上运行的客户端和服务器之间的通信。

● 远程通信：利用 SSE 与 HTTP 结合，实现跨网络的实时数据传输，适用于需要访问远程资源或分布式部署的场景。

现有MCP Server

有很多可以直接使用的MCP server

百度MCP 广场：

阿里

github项目

GitHub - punkpeye/awesome-mcp-servers: A collection of MCP servers.

有非常多的github上开源的mcp server

使用示例

以高德地图的mcp server为例。

mcp server配置文件（json）示例：

stdio

{
  "mcpServers": {
  #高德
    "amap-maps": {
      "command": "npx",
      "args": ["-y", "@amap/amap-maps-mcp-server"],
      "env": {
        "AMAP_MAPS_API_KEY": "您在高德官网上申请的key"
      }
    }
  }
  #另一个
  "sequential-thinking": {   
     "command": "npx",    
     "args": ["-y",       
             "@modelcontextprotocol/server-sequential-thinking"
       ]
   } 
   }
}

{"command": "npx",

"args": ["-y", "@amap/amap-maps-mcp-server"],

"env": {

"AMAP_MAPS_API_KEY": "您在高德官网上申请的key"

}

}

sse

{
  "mcpServers": {
    "amap-amap-sse": {
      "url": "https://mcp.amap.com/sse?key=您在高德官网上申请的key"
    }
  }
}

一个mcp server中可以有多个工具等服务。

基于adk的简单使用

# stdio
tools, exit_stack = await MCPToolset.from_server(
      connection_params=StdioServerParameters(
          command='npx',
          args=["-y",
                "@amap/amap-maps-mcp-server"
          ],
          env={
              "AMAP_MAPS_API_KEY": Amaps_api_key
          }
      )
  )
  
# sse
tools, exit_stack = await MCPToolset.from_server(
      connection_params=SseServerParams(
          url="http://0.0.0.0:9000/sse"
      ),
  )

FastMCP

文档：The FastMCP Server - FastMCP

虽然 FastMCP 仍然支持 SSE，但它已被弃用，Streamable HTTP 是新项目的首选。

构建MCP Server

from typing import Any
import httpx
from mcp.server.fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("weather")
# 工具 function
@mcp.tool()
async def get_alerts(state: str) -> str:
    """Get weather alerts for a US state.
    pass
# 可请求特定数据的参数化资源
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"
# 用于指导llm的可重用消息模板
@mcp.prompt()
def analyze_data(data_points: list[float]) -> str:
    """Creates a prompt asking for analysis of numerical data."""
    formatted_data = ", ".join(str(point) for point in data_points)
    return f"Please analyze these data points: {formatted_data}"

if __name__ == "__main__":
    # Initialize and run the server
    # mcp.run(transport='stdio')
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",
        port=4200,
        path="/my-custom-path",
        log_level="debug",
    )

MCP Client

# streamable-http传输类型
import asyncio
from fastmcp import Client
async def example():
    async with Client("http://127.0.0.1:4200/my-custom-path") as client:
        await client.list_tools()
if __name__ == "__main__":
    asyncio.run(example())