MCP协议应用实践：构建可扩展的AI服务接口
一、MCP协议核心概念
MCP（Model Context Protocol）是由Anthropic提出的一种开放标准，旨在打通大模型与外部系统之间的连接壁垒。它通过标准化的工具调用机制，使AI能够访问数据库、执行计算任务、调用外部API或操作本地文件，从而实现真正意义上的"智能代理"。
二、支持MCP的主流模型
当前具备稳定Function Calling能力的模型均能良好兼容MCP。推荐使用以下模型：
- DeepSeek V3（0324版本）
- Claude 3.7
- GPT-4o
- Gemini 2.5 Pro
可通过OpenRouter平台查询模型参数支持情况，若存在Tools字段则表示支持MCP。
三、开发环境配置
推荐使用UV作为包管理器：
# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh
# Windows
irm https://astral.sh/uv/install.ps1 | iex
# 或通过pip安装
pip install uv四、MCP通信模式与实现
MCP采用客户端-服务器架构，支持两种传输方式：
- stdio：基于进程间通信，适合本地脚本调用
- sse：基于HTTP的流式响应，适用于远程服务部署
两者均遵循JSON-RPC协议进行数据交互。
1. 基于stdio的本地服务实现
以下示例展示如何创建一个简单的加法计算服务：
from mcp.server.fastmcp import FastMCP
app = FastMCP("数学计算服务")
@app.tool()
def add_numbers(x: float, y: float) -> float:
"""执行两数相加运算"""
return x + y
if __name__ == '__main__':
app.run(transport='stdio')2. 客户端调用逻辑
客户端通过子进程启动服务，并动态解析可用工具：
from mcp.client.stdio import StdioServerParameters, stdio_client
from mcp import ClientSession
import asyncio
import json
from openai import OpenAI
class MCPPromptHandler:
def __init__(self, server_script: str):
self.client = OpenAI(
api_key="your_api_key",
base_url="https://api.deepseek.com"
)
self.server_script = server_script
async def execute(self, user_query: str):
params = StdioServerParameters(
command="python",
args=[self.server_script],
env=None
)
async with stdio_client(params) as (read_stream, write_stream):
async with ClientSession(read_stream=read_stream, write_stream=write_stream) as session:
await session.initialize()
tools_list = await session.list_tools()
# 构建符合Function Calling规范的工具列表
tool_specs = [
{
'type': 'function',
'function': {
'name': tool.name,
'description': tool.description,
'input_schema': tool.inputSchema
}
} for tool in tools_list.tools
]
messages = [{'role': 'user', 'content': user_query}]
response = self.client.chat.completions.create(
model="deepseek-chat",
messages=messages,
tools=tool_specs
)
choice = response.choices[0]
if choice.finish_reason == 'tool_calls':
messages.append(choice.message.model_dump())
call = choice.message.tool_calls[0]
func_name = call.function.name
args = json.loads(call.function.arguments)
result = await session.call_tool(name=func_name, arguments=args)
messages.append({
'role': 'tool',
'content': result.content[0].text,
'tool_call_id': call.id
})
final_response = self.client.chat.completions.create(
model="deepseek-chat",
messages=messages
)
print(f"结果：{final_response.choices[0].message.content}")
else:
print(f"直接回答：{choice.message.content}")
# 启动执行
asyncio.run(MCPPromptHandler("./math_server.py").execute("计算3.5和4.2的和"))3. 扩展功能：天气查询服务
在原有基础上添加天气查询能力：
WEATHER_DB = {
"福州": {"temp": 25, "condition": "晴朗", "wind": "微风", "humidity": 45},
"上海": {"temp": 28, "condition": "多云", "wind": "东风3级", "humidity": 60}
}
@app.tool()
def query_weather(city: str) -> str:
"""获取指定城市的实时天气信息"""
city = city.strip()
if city not in WEATHER_DB:
return f"不支持的城市，请选择：{', '.join(WEATHER_DB.keys())}"
data = WEATHER_DB[city]
return (f"{city}天气：{data['condition']}，"
f"温度{data['temp']}°C，"
f"风力{data['wind']}，"
f"湿度{data['humidity']}%")五、Java后端集成（基于Spring AI）
使用Spring Boot搭建MCP服务端：
application.yml配置
server:
port: 8080
spring:
ai:
mcp:
server:
name: weather-server
sse-message-endpoint: /mcp/message
version: 1.0.0
type: SYNC
instructions: "提供城市天气查询功能"
sse-endpoint: /sse
capabilities:
tool: true
resource: false
prompt: false
completion: falseWeatherService.java
@Service
public class WeatherService {
private static final Map<String, WeatherData> DATA = new HashMap<>();
static {
DATA.put("福州", new WeatherData(25, "晴朗", "微风", 45));
DATA.put("上海", new WeatherData(28, "多云", "东风3级", 60));
}
@Tool(name = "getWeather", description = "根据城市名称获取今日天气")
public String getWeather(@ToolParam("城市名") String city) {
if (city == null || city.isEmpty()) {
return "请输入有效城市名称";
}
var data = DATA.get(city.trim());
if (data == null) {
return "暂无该城市天气数据，支持城市：" + DATA.keySet();
}
return String.format("【%s】气温：%d°C，天气：%s，风力：%s，湿度：%d%%",
city, data.temp, data.condition, data.wind, data.humidity);
}
private record WeatherData(int temp, String condition, String wind, int humidity) {}
}工具注册配置
@Configuration
public class ToolConfig {
@Bean
public ToolCallbackProvider weatherTools(WeatherService service) {
return MethodToolCallbackProvider.builder().toolObjects(service).build();
}
}客户端调用更新
if __name__ == '__main__':
client = MCPClient(server_url="http://127.0.0.1:8080/sse")
asyncio.run(client.run('福州今天天气如何'))