MCP 是什么
MCP 是一个开放协议，它为应用程序向 LLM 提供上下文的方式进行了标准化。
使用 MCP，像 Claude 或 ChatGPT 这样的 AI 应用程序可以连接到数据源（例如本地文件、数据库）、工具（例如搜索引擎、计算器）和工作流（例如专用提示）——使它们能够访问信息并执行任务。
你可以将 MCP 想象成 AI 应用程序的 USB-C 接口。就像 USB-C 为设备连接各种外设和配件提供了标准化的方式一样，MCP 为 AI 模型连接各种数据源和工具提供了标准化的接口。
可以用几个词来总结 MCP 的关键点：
开放，协议，标准化，接口，工具，工作流
为什么需要 MCP
为了提高 LLM 的能力边界，需要让它们能够访问外部数据和工具。然而，不同的 AI 应用程序通常使用不同的集成方法，这导致了碎片化和重复工作。
在没有 MCP 之前，每个 AI 应用都需要为不同的数据源和工具开发专门的集成方案。这导致了大量重复开发工作，缺乏互操作性，且维护成本高昂。MCP 的出现解决了这些问题，提供了一个统一的标准，让开发者可以一次开发，到处使用，同时降低项目中组件的耦合度。
怎么使用 MCP
MCP 的实现方式主要分为两种：自建远程服务和使用社区托管服务。
远程 MCP
当你需要在不同服务器之间进行 MCP 通信时，可以部署远程 MCP 服务器。这种方式适合企业级应用或需要自定义工具的场景。
- 服务端
import asyncio
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import TextContent, Tool
from starlette.applications import Starlette
from starlette.responses import Response
from starlette.routing import Mount, Route
app = Server("remote_tool_server")
@app.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="add_numbers",
description="Add two numbers together",
inputSchema={
"type": "object",
"properties": {
"a": {
"type": "number",
"description": "First number"
},
"b": {
"type": "number",
"description": "Second number"
}
},
"required": ["a", "b"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "add_numbers":
a = arguments.get("a", 0)
b = arguments.get("b", 0)
result = a + b
return [TextContent(type="text", text=f"The sum of {a} and {b} is {result}")]
return [TextContent(type="text", text="Unknown tool")]
# Create SSE transport
sse = SseServerTransport("/messages/")
async def handle_sse(request):
async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
await app.run(streams[0], streams[1], app.create_initialization_options())
return Response()
# Create Starlette app
starlette_app = Starlette(
routes=[
Route("/sse", endpoint=handle_sse, methods=["GET"]),
Mount("/messages/", app=sse.handle_post_message),
]
)
if __name__ == "__main__":
import uvicorn
print("Starting HTTP MCP server on port 8000...")
uvicorn.run(starlette_app, host="0.0.0.0", port=8000)
上面的服务器端代码创建了一个提供加法运算的 MCP 服务器，使用 SSE（Server-Sent Events）作为传输协议。服务器暴露了一个 add_numbers 工具，可以接收两个数字并返回它们的和。
- 客户端
import asyncio
import sys
from mcp.client.sse import sse_client
from mcp.client.session import ClientSession
import openai
# Custom base_url and model - replace with your actual values
BASE_URL = "api_base_url"
MODEL = "model"
API_KEY = "your_key" # Replace with your actual API key
client = openai.OpenAI(base_url=BASE_URL, api_key=API_KEY)
async def main():
user_query = "What is 15 + 27?"
# Connect to remote SSE MCP server
server_url = "http://localhost:8000/sse" # Adjust URL for remote server
async with sse_client(server_url) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# Get tools from server
tools_result = await session.list_tools()
tools = tools_result.tools
# Convert MCP tools to OpenAI format
openai_tools = []
for tool in tools:
openai_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
})
# Call LLM with tools
response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": user_query}],
tools=openai_tools,
tool_choice="auto"
)
message = response.choices[0].message
if message.tool_calls:
# LLM decided to call a tool
tool_call = message.tool_calls[0]
tool_name = tool_call.function.name
arguments = eval(tool_call.function.arguments) # In production, use json.loads
# Call the actual tool via MCP server
result = await session.call_tool(tool_name, arguments)
tool_result = result.content[0].text
# Send result back to LLM for final response
final_response = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "user", "content": user_query},
message,
{"role": "tool", "tool_call_id": tool_call.id, "content": tool_result}
]
)
print("Final response:", final_response.choices[0].message.content)
else:
# No tool call needed
print("LLM response:", message.content)
if __name__ == "__main__":
asyncio.run(main())
客户端代码连接到远程 MCP 服务器后，获取可用工具，并与 LLM 集成。当用户提出数学问题时，LLM 会智能地决定是否需要调用加法工具来解决问题。
但是现在的 LLM 普遍可以精准的直接回答两数之和是多少的问题，所以我们需要一个更复杂的例子来展示 MCP 的优势。
社区 MCP
除了自建服务器，现在也有很多社区托管的 MCP 服务，可以直接使用，无需自己搭建服务器。
例如 ModelScope 的 MCP 广场提供了丰富的预构建服务，涵盖了网页抓取、数据处理、API 调用等常见需求。以 fetch 服务为例，你可以直接使用远程 MCP，而无需自己搭建服务器。
以下就是一个使用 ModelScope 提供的 fetch 服务来抓取网页内容的示例：
import asyncio
import json
from mcp.client.sse import sse_client
from mcp.client.session import ClientSession
import openai
# Custom base_url and model - replace with your actual values
BASE_URL = "api_base_url"
MODEL = "model"
API_KEY = "your_key" # Replace with your actual API key
client = openai.OpenAI(base_url=BASE_URL, api_key=API_KEY)
async def main():
# 用户查询
user_query = "请帮我获取 https://www.python.org 的内容摘要"
# 替换为您的远程 MCP 服务器地址
# 这里使用 ModelScope 提供的 fetch 服务器地址（需要替换为实际地址）
server_url = "https://mcp.api-inference.modelscope.net/your_id/sse" # 替换为实际的远程服务器URL
try:
# 连接到远程 MCP 服务器
async with sse_client(server_url) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
print("已连接到远程 MCP 服务器")
# 获取可用工具
tools_result = await session.list_tools()
tools = tools_result.tools
print(f"可用工具: {[tool.name for tool in tools]}")
# 转换为 OpenAI 工具格式
openai_tools = []
for tool in tools:
openai_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
})
# 调用 LLM，让它决定是否需要使用工具
response = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": user_query}],
tools=openai_tools,
tool_choice="auto"
)
message = response.choices[0].message
if message.tool_calls:
# LLM 决定使用工具
print("LLM 决定使用网页抓取工具...")
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f"调用工具: {tool_name}")
print(f"参数: {arguments}")
# 通过 MCP 调用远程工具
result = await session.call_tool(tool_name, arguments)
tool_result = result.content[0].text
print(f"工具结果长度: {len(tool_result)} 字符")
# 将工具结果发送回 LLM 获取最终响应
final_response = client.chat.completions.create(
model=MODEL,
messages=[
{"role": "user", "content": user_query},
message,
{"role": "tool", "tool_call_id": tool_call.id, "content": tool_result}
]
)
print("\n=== 最终响应 ===")
print(final_response.choices[0].message.content)
else:
# LLM 不需要使用工具
print("LLM 响应:", message.content)
except Exception as e:
print(f"连接远程 MCP 服务器失败: {e}")
print("\n作为演示，这里使用本地 fetch 工具:")
if __name__ == "__main__":
asyncio.run(main())
其他
简单的说，这两张图我认为就是 MCP 的精髓
评论区