我要投稿

MCP 如何成为 AI Agent 的“USB-C 接口”？附搭建教程

发布日期：2025-06-05 08:28:27 浏览次数： 1577 作者：TalkJava

MCP Explained: The New Standard Connecting AI to Everything | by EdwinLisowski | Apr, 2025 | Medium" class="rich_pages wxw-img" data-ratio="0.5888888888888889" data-type="png" data-w="1080" data-imgfileid="100001488">

什么是 MCP？

MCP（模型上下文协议，Model Context Protocol）是一种开放标准，旨在标准化大型语言模型（LLM）与外部工具、数据源之间的通信方式。它采用客户端-服务器架构，支持多种通信协议和传输机制，以实现结构化、多轮次、可扩展的上下文交换。就像 AI 应用程序的 USB-C 端口一样，提供了一种标准化的方式将 AI 模型连接到不同的数据源和工具。

在 MCP 出现之前，开发人员必须为每个 AI 应用程序所需的数据源或工具构建自定义连接——这是一个非常耗时且重复的过程。

MCP 采用客户端-服务器架构，AI 模型作为客户端，通过 MCP 协议与 MCP 服务器进行通信。MCP 服务器负责与外部数据源或工具交互，并将获取的数据按照 MCP 协议规范格式化后返回给客户端。这种设计使得 AI 模型能够动态地获取所需的上下文信息，执行更广泛的任务。大大缩短了开发时间，并降低开发复杂性。

MCP 架构

MCP 遵循 客户端-服务器 架构，其中：

主机
（Host）是发起连接的 LLM 应用程序（如 Claude Desktop 或 IDE）。
客户端
（MCP Client）在主机应用程序内部与服务器保持 1：1 连接。
服务器
（MCP Server）向客户提供上下文、工具 Tools 和提示 Prompts。

MCP 支持三种功能：

工具（Tools）
可被 AI 调用的函数（如发送邮件、查询天气）。
资源（Resources）
可读取的数据（如文件、数据库记录）。
提示（Prompts）
预设的指令模板，优化任务执行。

MCP 传输通信

MCP 的通信基于 JSON-RPC 2.0 协议，这是一种轻量级的远程过程调用协议，使用 JSON 格式进行数据交换。该协议支持状态化连接，允许在一个会话中进行多次请求和响应，适用于多轮对话和复杂的工具调用流程。

MCP 支持以下两种主要的传输机制：

标准输入/输出（Stdio）
• 适用场景：客户端和服务器在同一台机器上运行的本地集成。
• 特点：使用标准输入输出流进行通信，适合访问本地文件系统或执行本地脚本的场景。
HTTP + Server-Sent Events（SSE）
• 适用场景：客户端和服务器分布式部署的远程通信。
• 特点：客户端通过 HTTP POST 向服务器发送请求，服务器通过 SSE 向客户端推送实时消息，支持实时数据流和事件驱动的通信。

在 MCP 架构中，通信流程如下： 1. 客户端：位于主机应用（如聊天机器人、IDE 助手）中，负责构建请求并发送给 MCP 服务器。

服务器：提供外部数据源或工具的访问接口，接收客户端的请求，处理后返回结构化的响应。
通信协议：客户端和服务器之间通过 JSON-RPC 2.0 协议进行通信，传输机制可以是 Stdio 或 HTTP + SSE。

MCP 提供了一个统一的通信协议和多样的传输机制，支持大型语言模型与外部工具和数据源之间的高效集成。通过标准化的结构和灵活的传输方式，MCP 使得开发者能够构建更强大、可扩展的 AI 应用。

MCP 与 Function Calling（Tool Call）

Function Calling（函数调用）是模型内部的函数调用机制，是由 LLM 提供商（如 OpenAI、Anthropic）实现的一种机制，允许模型根据用户输入，生成结构化的函数调用请求。 不同平台之间可能存在差异。 适合处理边界清晰、描述明确的任务，如数据提取、分类或外部 API 调用等。代码的适配性和复用性较差（我们需要把每个 Function 编码到程序中）。

而 MCP（模型上下文协议） 是模型与外部系统的通用通信协议。使得不同的 AI 模型和外部系统能够无缝集成，降低了适配成本。更擅长处理复杂、多步骤的对话场景，尤其是在需要维持上下文连贯性和动态适应用户需求的场景中，其优势尤为明显。

MCP 实战

1. 本地搭建天气 MCP Server & Client

构建 MCP Weather Server


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

# 初始化 FastMCP server
mcp = FastMCP("weather")

定义 MCP Server tools


# 获取美国某个州的天气警报
@mcp.tool()
async def get_alerts(state: str)-> str:
"""Get weather alerts for a US state.

Args:
state:Two-letter US state code(e.g.CA,NY)
"""
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data =awaitmake_nws_request(url)

if not data or "features" not indata:
return"Unable to fetch alerts or no alerts found."

if not data["features"]:
return"No active alerts for this state."

    alerts =[format_alert(feature)for feature in data["features"]]
return"\n---\n".join(alerts)


# 获取某个地点的天气警报
@mcp.tool()
async def get_forecast(latitude: float,longitude: float)-> str:
"""Get weather forecast for a location.

Args:
latitude:Latitudeof the location
longitude:Longitudeof the location
"""
    # Firstget the forecast grid endpoint
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data =awaitmake_nws_request(points_url)

if not points_data:
return"Unable to fetch forecast data for this location."

    # Get the forecast URLfrom the points response
    forecast_url = points_data["properties"]["forecast"]
    forecast_data =awaitmake_nws_request(forecast_url)

if not forecast_data:
return"Unable to fetch detailed forecast."

    # Format the periods into a readable forecast
    periods = forecast_data["properties"]["periods"]
    forecasts =[]
for period in periods[:5]:  # Only show next 5 periods
        forecast = f"""
{period['name']}:
Temperature:{period['temperature']}°{period['temperatureUnit']}
Wind:{period['windSpeed']}{period['windDirection']}
Forecast:{period['detailedForecast']}
"""
        forecasts.append(forecast)

return"\n---\n".join(forecasts)

使用 main 方法启动 Weather MCP Server


if __name__ =="__main__":
    # Initialize and run the server
    mcp.run(transport='stdio')

定义 Weather MCP Client Class：


importasyncio
from typing importOptional
from contextlib importAsyncExitStack

from mcp importClientSession,StdioServerParameters
from mcp.client.stdioimportstdio_client

from anthropic importAnthropic
from dotenv import load_dotenv

load_dotenv()  # load environment variables from.env

classMCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session:Optional[ClientSession]=None
        self.exit_stack=AsyncExitStack()
        self.anthropic=Anthropic()
    # methods will go here

连接 MCP Server


async def connect_to_server(self,server_script_path: str):
"""Connect to an MCP server

Args:
server_script_path:Path to the server script(.py or .js)
"""
    is_python = server_script_path.endswith('.py')
    is_js = server_script_path.endswith('.js')
ifnot(is_python or is_js):
        raise ValueError("Server script must be a .py or .js file")

    command ="python"if is_python else"node"
    server_params =StdioServerParameters(
        command=command,
        args=[server_script_path],
        env=None
)

    stdio_transport =await self.exit_stack.enter_async_context(stdio_client(server_params))
    self.stdio, self.write= stdio_transport
    self.session=await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))

await self.session.initialize()

    # List available tools
    response =await self.session.list_tools()
    tools = response.tools
print("\nConnected to server with tools:",[tool.namefor tool in tools])

查询处理逻辑


async def process_query(self,query: str)-> str:
"""Process a query using Claude and available tools"""
    messages =[
{
"role":"user",
"content": query
}
]

    response =await self.session.list_tools()
    available_tools =[{
"name": tool.name,
"description": tool.description,
"input_schema": tool.inputSchema
}for tool in response.tools]

    # InitialClaudeAPI call
    response = self.anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
)

    # Process response and handle tool calls
    final_text =[]

    assistant_message_content =[]
for content in response.content:
if content.type'text':
            final_text.append(content.text)
            assistant_message_content.append(content)
        elif content.type'tool_use':
            tool_name = content.name
            tool_args = content.input

            # Execute tool call
            result =await self.session.call_tool(tool_name, tool_args)
            final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")

            assistant_message_content.append(content)
            messages.append({
"role":"assistant",
"content": assistant_message_content
})
            messages.append({
"role":"user",
"content":[
{
"type":"tool_result",
"tool_use_id": content.id,
"content": result.content
}
]
})

            # Get next response fromClaude
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
)

            final_text.append(response.content[0].text)

return"\n".join(final_text)

注意：不同 LLM 的 request 和 response 的结构是不一样的，DeepSeek 和千问模型遵循 OpenAI 的规范，而 Claude 与 Gemini 的 resposne 与 openai 不一样。如：

Claude response.content[0].text 与 OpenAI response.choices[0].message.content

交互式聊天界面


async def chat_loop(self):
"""Run an interactive chat loop"""
    print("\nMCP Client Started!")
    print("Type your queries or 'quit' to exit.")

while True:
        try:
            query = input("\nQuery: ").strip()

if query.lower()=='quit':
break

            response = await self.process_query(query)
            print("\n" + response)

        except Exception as e:
            print(f"\nError: {str(e)}")

async def cleanup(self):
"""Clean up resources"""
    await self.exit_stack.aclose()

测试


async def main():
iflen(sys.argv)<2:
print("Usage: python client.py <path_to_server_script>")
        sys.exit(1)

    client =MCPClient()
try:
await client.connect_to_server(sys.argv[1])
await client.chat_loop()
finally:
await client.cleanup()

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

启动 Client



python client.py./weather.py

# Relative path
uv run client.py./server/weather.py

# Absolute path
uv run client.py/Users/username/projects/mcp-server/weather.py

# Windowspath(either format works)
uv run client.pyC:/projects/mcp-server/weather.py
uv run client.pyC:\\projects\\mcp-server\\weather.py

2.启动 GitHub MCP Server

MCP Server Github提供很多现成的MCP Server，GitHub MCP Server 是一个模型上下文协议 (MCP) 服务器，可与 GitHub API 无缝集成，为开发人员和工具提供高级自动化和交互功能。

使用 npx 直接运行 GitHub MCP Server。需要创建一个 GitHub 个人访问令牌

（https://github.com/settings/tokens）



npx -y @modelcontextprotocol/server-github

使用官方提供的 MCP Inspector 工具来测试和调试 MCP Server。


npx @modelcontextprotocol/inspector npx -y @modelcontextprotocol/server-github

浏览器访问 http://localhost:5173

3.LLM 集成 GitHub MCP Server

LLM 集成 MCP Server 有很多种方式，可以自定义，也可以直接使用已有的库。

使用mcp-use 进行 LLM 与 MCP 的集成，mcp-use是一个开源的 Python 库，可以非常轻松地将任何 LLM 连接到本地和远程的任何 MCP 服务器。

下面代码使用Deepseek LLM 集成Github MCP Server：


import asyncio
importos
from dotenv importload_dotenv
from langchain_openai importChatOpenAI
from mcp_use importMCPAgent,MCPClient
"""
pip install mcp-use
"""
async def main():
    # Load environment variables
load_dotenv()

    # Create configuration dictionary
    config ={
"mcpServers":{
"github":{
"command":"npx",
"args":["-y","@modelcontextprotocol/server-github"],
"env":{
"GITHUB_PERSONAL_ACCESS_TOKEN":"<GITHUB PERSONAL ACCESS TOKEN>"
      }
    }
 }
}

    # CreateMCPClientfrom configuration dictionary
    client =MCPClient.from_dict(config)

    # CreateLLM
    llm =ChatOpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"),
                     base_url="https://api.deepseek.com",
                     model="deepseek-chat")

    # Create agent with the client
    agent =MCPAgent(llm=llm, client=client, max_steps=30)

    # Run the query
    result =await agent.run(
"search ai-agent-demo repo",
)
print(f"\nResult: {result}")

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