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（模型上下文协议，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
2.启动 GitHub MCP Server
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
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
LLM 集成 MCP Server 有很多种方式，可以自定义，也可以直接使用已有的库。
使用mcp-use 进行 LLM 与 MCP 的集成，mcp-use是一个开源的 Python 库，可以非常轻松地将任何 LLM 连接到本地和远程的任何 MCP 服务器。
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())
