MCP 原理解析與效果實測
MCP 架構簡述
MCP 是由 Anthropic 推出的開源協議,目的是通過統一的連接方式,讓大型語言模型(LLM)與外部數據源和工具無縫集成,減少重複造輪子的過程。
MCP 架構主要涉及到以下 5 個部分:
● Host:一個包含 MCP Client 的應用,可以是 Web、App、或其他類型的程序等
● MCP Client:使用 MCP 協議與 Server 建立一對一連接
● MCP Server:連接內部、外部、網絡資源,使用 MCP 協議對外提供服務
● Local:內部資源
● Remote:外部 / 網絡資源
OpenAI 提出的function calling
增強了大模型的能力,通過使用外部工具的方式提升了模型的智能化水平。但是不同的模型使用工具的方式和定義並不完全相同,而且,對於常見的功能需求,比如聯網查詢和天氣查詢,也都需要在自己的程序中實現相應的代碼。而有了 MCP 之後,只需要把這些已經被實現的功能引入進自己的項目中,就像導入一個包一樣簡單。
因此,從實際體驗上來講,可以簡單理解成 MCP 是把 Agent 涉及到的工具調用、提示詞等進行了標準化處理,簡化了開發流程,減少了重複造輪子的國產,方便大家無縫使用開源社區中優秀的工具。
Cursor MCP 實戰
在網上找了一個 MCP Server,用於檢索 arxiv 文章(鏈接:https://github.com/blazickjp/arxiv-mcp-server),這裏我使用 Cursor 作爲 MCP Client 使用,不過需要注意 Cursor 僅支持 MCP 中的 tools。
- 安裝對應的 Server
uv tool install arxiv-mcp-server
- Cursor 配置
在「Cursor Settings」菜單欄中點擊「MCP」,然後點擊「+ Add new global MCP server」按鈕,會彈出一個 mcp.json 文件,把 arxiv-mcp-server 中的配置信息添加到 json 文件中去。
{
"mcpServers":{
"arxiv-mcp-server":{
"command":"uv",
"args":[
"tool",
"run",
"arxiv-mcp-server",
"--storage-path","/path/to/paper/storage"
]
}
}
}
當該 MCP Server 前面的按鈕顯示綠色的時候,說明該功能已經可以正常使用了。
- 效果實測
在這裏我輸入一個問題「ai agent 2025」,針對改問題,模型判斷最適合的工具應該是聯網搜索功能,由於這裏我點擊了拒絕,因此模型判斷應該選擇工具「search_papers」,可以看到顯示的工具名稱正是之前配置的 MCP Server 中的工具名稱,因此這裏可以選擇使用該工具繼續。這裏可以看到 search_papers 的結果:
Parameters:
{
"query":"AI agents autonomous systems 2025",
"categories":[
"cs.AI",
"cs.LG"
],
"max_results":5
}
Result:
{
"total_results":5,
"papers":[
{
"id":"2503.10638v1",
"title":"Studying Classifier(-Free) Guidance From a Classifier-Centric Perspective",
"authors":[
"Xiaoming Zhao",
"Alexander G. Schwing"
],
"abstract":"Classifier-free guidance has become a staple for conditional generation with\ndenoising diffusion models. However, a comprehensive understanding of\nclassifier-free guidance is still missing. In this work, we carry out an\nempirical study to provide a fresh perspective on classifier-free guidance.\nConcretely, instead of solely focusing on classifier-free guidance, we trace\nback to the root, i.e., classifier guidance, pinpoint the key assumption for\nthe derivation, and conduct a systematic study to understand the role of the\nclassifier. We find that both classifier guidance and classifier-free guidance\nachieve conditional generation by pushing the denoising diffusion trajectories\naway from decision boundaries, i.e., areas where conditional information is\nusually entangled and is hard to learn. Based on this classifier-centric\nunderstanding, we propose a generic postprocessing step built upon\nflow-matching to shrink the gap between the learned distribution for a\npre-trained denoising diffusion model and the real data distribution, majorly\naround the decision boundaries. Experiments on various datasets verify the\neffectiveness of the proposed approach.",
"categories":[
"cs.CV",
"cs.AI",
"cs.LG"
],
"published":"2025-03-13T17:59:59+00:00",
"url":"http://arxiv.org/pdf/2503.10638v1",
"resource_uri":"arxiv://2503.10638v1"
},
{
"id":"2503.10636v1",
"title":"The Curse of Conditions: Analyzing and Improving Optimal Transport for Conditional Flow-Based Generation",
"authors":[
"Ho Kei Cheng",
"Alexander Schwing"
],
"abstract":"Minibatch optimal transport coupling straightens paths in unconditional flow\nmatching. This leads to computationally less demanding inference as fewer\nintegration steps and less complex numerical solvers can be employed when\nnumerically solving an ordinary differential equation at test time. However, in\nthe conditional setting, minibatch optimal transport falls short. This is\nbecause the default optimal transport mapping disregards conditions, resulting\nin a conditionally skewed prior distribution during training. In contrast, at\ntest time, we have no access to the skewed prior, and instead sample from the\nfull, unbiased prior distribution. This gap between training and testing leads\nto a subpar performance. To bridge this gap, we propose conditional optimal\ntransport C^2OT that adds a conditional weighting term in the cost matrix when\ncomputing the optimal transport assignment. Experiments demonstrate that this\nsimple fix works with both discrete and continuous conditions in\n8gaussians-to-moons, CIFAR-10, ImageNet-32x32, and ImageNet-256x256. Our method\nperforms better overall compared to the existing baselines across different\nfunction evaluation budgets. Code is available at\nhttps://hkchengrex.github.io/C2OT",
"categories":[
"cs.LG",
"cs.CV"
],
"published":"2025-03-13T17:59:56+00:00",
"url":"http://arxiv.org/pdf/2503.10636v1",
"resource_uri":"arxiv://2503.10636v1"
}
到這裏模型判斷已經解決了用戶的問題,因此沒有下一步。 接下來的交互中,我明確要求模型下載其中的一篇論文,可以看到工具「download_papers」已經被調用,pdf 論文被保存爲. md 格式,並存放在本地路徑中。然而還沒有結束,模型會接着調用工具「read_papers」,針對論文內容進行解讀,直到這一步才終止。
自定義 MCP Client 和 MCP Server
前面提到的 Cursor 相當於 MCP 架構中的 Client(實際上是包含了 Client 的 host),因此對於普通用戶來說,只需要關注自身需要什麼樣的功能(Server),去找到相應的功能(Server)安裝即可。對於開發者而言,如果需要讓自己的程序能夠使用已有的 MCP Server,則需要針對自己的程序進行改造,使其符合 MCP 規範纔行。目前官方也分別提供了 Python 和 Js 的 SDK,可以很方便進行開發。
- MCP Client
根據官方提供的教程:
https://github.com/modelcontextprotocol/quickstart-resources/tree/main/mcp-client-python
首先,創建一個連接 Server 的函數,在該函數中,判斷腳本類型是否滿足(只允許 python 和 js),並且返回可以使用的工具。
async defconnect_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')
if not(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.name for tool in tools])
接下來,定義輸入解析函數,這一步就是大模型工具調用的處理流程。
async defprocess_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]
# Initial Claude API 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 =[]
for content in response.content:
if content.type=='text':
final_text.append(content.text)
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}]")
# Continue conversation with tool results
if hasattr(content,'text') and content.text:
messages.append({
"role":"assistant",
"content": content.text
})
messages.append({
"role":"user",
"content": result.content
})
# Get next response from Claude
response = self.anthropic.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=1000,
messages=messages,
)
final_text.append(response.content[0].text)
return"\n".join(final_text)
處理對話函數,這裏的 demo 不保留歷史記錄,只支持單次對話。
async defchat_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 Exceptionas e:
print(f"\nError: {str(e)}")
官方的例子中用到的是 Claude 模型,但是實測可以兼容 OpenAI 接口的 API,只要能支持tools
這個參數即可。
- MCP Server
MCP Server 主要包含了以下 3 部分的內容:
● Resources:允許訪問的靜態資源,如文件圖片等
● Tools:該 Server 擁有的工具
● Prompts:預先寫好的提示詞模版,輔助完成特定任務
官方也提供了一個簡單的 Server 開發 demo:
https://github.com/modelcontextprotocol/quickstart-resources/tree/main/weather-server-python
首先,定義了 2 個函數,用於發起實際的 API 請求以及解析響應內容。
async defmake_nws_request(url: str)->dict[str,Any]|None:
"""Make a request to the NWS API with proper error handling."""
headers ={
"User-Agent": USER_AGENT,
"Accept":"application/geo+json"
}
async with httpx.AsyncClient()as client:
try:
response =await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
return response.json()
except Exception:
returnNone
def format_alert(feature: dict)->str:
"""Format an alert feature into a readable string."""
props = feature["properties"]
return f"""
Event: {props.get('event', 'Unknown')}
Area: {props.get('areaDesc', 'Unknown')}
Severity: {props.get('severity', 'Unknown')}
Description: {props.get('description', 'No description available')}
Instructions: {props.get('instruction', 'No specific instructions provided')}
"""
定義了 2 個工具函數,通過添加裝飾器讓 Client 知道該函數是工具,可以進行調用。
@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 = await make_nws_request(url)
if not data or "features" not in data:
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: Latitude of the location
longitude: Longitude of the location
"""
# First get the forecast grid endpoint
points_url =f"{NWS_API_BASE}/points/{latitude},{longitude}"
points_data =await make_nws_request(points_url)
if not points_data:
return"Unable to fetch forecast data for this location."
# Get the forecast URL from the points response
forecast_url = points_data["properties"]["forecast"]
forecast_data =await make_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)
好用的 MCP 推薦
MCP Server:
https://github.com/punkpeye/awesome-mcp-servers/
MCP Client:
https://github.com/punkpeye/awesome-mcp-clients/
也可以在 glama 中根據條件篩選:
https://glama.ai/mcp/servers
總結
MCP 的出現,對於整個 AI 生態都起到了促進作用:
普通用戶只需要關注自己需要什麼樣的插件,然後去找,並且安裝到自己的 Cursor 等 AI 應用上即可使用; 開發者根據自身項目需要,可以通過引入優秀的 MCP Server 來減少開發的工作量,也可以把重複實現的功能作爲 MCP Server 進行開發,進而實現多項目複用。
不過,Claude 模型及其桌面端在國內均無法直接使用,需要外國手機號才能註冊,並且經常有被 Ban 的風險;而 Cursor 在 Agent 模式下才能使用 MCP Server,普通 Chat 模式是不含該功能的,並且 Agent 模式在免費版下每個月僅有一定額度。這也只能寄希望於開源社區湧現出更多優秀的項目,才能發揮出 MCP 生態的影響。
你好,我是 William,一名喜歡折騰的程序員。曾擔任大廠算法工程師,現在初創公司做 AI 應用全棧開發。
工作內容與興趣點集中於大模型、RAG、文檔處理等方向,平常也會介紹相關的開源工作,歡迎圍觀。
本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源:https://mp.weixin.qq.com/s/w7v4NuaIeZGeThMA8p8MeA