Function Calling 下一步：MCP 协议与 AI 工具生态

Function Calling 下一步：MCP 协议与 AI 工具生态

一、Function Calling 的现状与局限

Function Calling（函数调用）是大语言模型（LLM）与外部系统交互的核心机制。自 2023 年 OpenAI 推出 Function Calling 功能以来，这一能力迅速成为 AI Agent 开发的基础。通过 Function Calling，LLM 可以根据用户输入生成结构化的函数调用参数，从而查询数据库、调用 API、执行代码等操作。

当前 Function Calling 的实现方式：

现有方案的局限性：

尽管 Function Calling 已经广泛应用，但仍然存在以下局限：

# 当前 Function Calling 的碎片化问题示例

# OpenAI Function Calling 格式
openai_tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "获取指定城市的天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["city"]
}
}
}
]

# LangChain Tool 格式
from langchain.tools import Tool

def get_weather(city: str, unit: str = "celsius") -> str:
"""获取指定城市的天气"""
# 实现逻辑…
pass

langchain_tool = Tool(
name="get_weather",
func=get_weather,
description="获取指定城市的天气"
)

# Semantic Kernel Skill 格式
import semantic_kernel as sk

skill = sk.Skill("WeatherSkill")
@skill.semantic_function("获取指定城市的天气")
def get_weather(city: str, unit: str = "celsius") -> str:
# 实现逻辑…
pass

# 问题：同一功能需要为不同框架编写多次，无法复用

工具定义的碎片化严重阻碍了 AI 工具生态的发展。正如 Web 服务需要统一的 API 标准（REST、GraphQL），AI 工具也需要统一的协议来实现互操作性。这就是 MCP（Model Context Protocol）诞生的背景。

二、MCP 协议：AI 工具的"USB 接口"

MCP（Model Context Protocol）是由 Anthropic 于 2024 年底提出的开放协议，旨在统一 AI 应用与外部工具、数据源之间的交互方式。MCP 借鉴了 Language Server Protocol（LSP）的设计理念，通过定义标准的客户端-服务器架构，使得任何 AI 应用（Client）都可以动态发现和调用任何工具提供方（Server）提供的功能。

MCP 的核心设计理念：

MCP 的工作流程：

# MCP 协议示例：实现一个简单的 MCP Server

from mcp import Server, Tool
from mcp.server.stdio import stdio_server
from pydantic import BaseModel, Field
import httpx
import json

# 定义工具输入 Schema
class GetWeatherInput(BaseModel):
"""获取天气工具的输入参数"""
city: str = Field(description="城市名称，如 '北京'、'Shanghai'")
unit: str = Field(default="celsius", description="温度单位：celsius 或 fahrenheit")

# 创建 MCP Server
server = Server("weather-server")

@server.tool(
name="get_weather",
description="获取指定城市的当前天气信息",
input_schema=GetWeatherInput
)
async def get_weather(city: str, unit: str = "celsius") -> dict:
"""
获取指定城市的天气

Args:
city: 城市名称
unit: 温度单位

Returns:
包含天气信息的字典
"""
# 调用天气 API（示例）
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://api.weather.com/v1/weather",
params={"city": city, "unit": unit},
headers={"Authorization": f"Bearer {self.api_key}"}
)
data = response.json()

return {
"city": city,
"temperature": data["temp"],
"unit": unit,
"condition": data["condition"],
"humidity": data["humidity"]
}

@server.tool(
name="get_forecast",
description="获取指定城市未来几天的天气预报"
)
async def get_forecast(city: str, days: int = 3) -> dict:
"""获取天气预报"""
# 实现逻辑…
pass

# 启动 Server（使用 stdio 传输）
if __name__ == "__main__":
import asyncio
asyncio.run(stdio_server(server))

# MCP Client 示例：在 AI 应用中调用 MCP Server

from mcp import Client
from mcp.client.stdio import stdio_client
import anthropic

class MCPEnabledAgent:
"""支持 MCP 的 AI Agent"""

def __init__(self, mcp_server_command: list):
# 启动 MCP Server 进程
self.mcp_client = stdio_client(mcp_server_command)

# 初始化 Anthropic 客户端
self.anthropic_client = anthropic.Anthropic()

# 工具缓存
self.available_tools = None

async def initialize(self):
"""初始化：连接 MCP Server 并获取工具列表"""
# 连接到 Server
await self.mcp_client.initialize()

# 获取工具列表
response = await self.mcp_client.list_tools()
self.available_tools = response.tools

# 转换为 Anthropic Function Calling 格式
self.anthropic_tools = self._convert_to_anthropic_format(self.available_tools)

print(f"已加载 {len(self.available_tools)} 个工具")
for tool in self.available_tools:
print(f" – {tool.name}: {tool.description}")

def _convert_to_anthropic_format(self, mcp_tools: list) -> list:
"""将 MCP 工具格式转换为 Anthropic Function Calling 格式"""
anthropic_tools = []

for tool in mcp_tools:
anthropic_tool = {
"name": tool.name,
"description": tool.description,
"parameters": tool.input_schema
}
anthropic_tools.append(anthropic_tool)

return anthropic_tools

async def chat(self, user_message: str) -> str:
"""处理用户消息，支持工具调用"""

# 调用 LLM
response = self.anthropic_client.messages.create(
model="claude-3-sonnet-20240229",
max_tokens=4096,
tools=self.anthropic_tools,
messages=[{"role": "user", "content": user_message}]
)

# 检查是否需要调用工具
if response.stop_reason == "tool_use":
# 提取工具调用请求
tool_use_blocks = [block for block in response.content if block.type == "tool_use"]

tool_results = []
for block in tool_use_blocks:
tool_name = block.name
tool_input = block.input

print(f"调用工具: {tool_name}, 参数: {tool_input}")

# 通过 MCP 调用工具
result = await self.mcp_client.call_tool(tool_name, tool_input)

tool_results.append({
"tool_use_id": block.id,
"content": result.content
})

# 将工具执行结果返回给 LLM
final_response = self.anthropic_client.messages.create(
model="claude-3-sonnet-20240229",
max_tokens=4096,
messages=[
{"role": "user", "content": user_message},
{"role": "assistant", "content": response.content},
{"role": "user", "content": tool_results}
]
)

return final_response.content[0].text

else:
# 无需调用工具，直接返回响应
return response.content[0].text

async def close(self):
"""关闭连接"""
await self.mcp_client.close()

# 使用示例
async def main():
# 启动 MCP Server（假设 Server 可执行文件为 mcp_weather_server）
agent = MCPEnabledAgent(mcp_server_command=["python", "mcp_weather_server.py"])

await agent.initialize()

# 用户交互
response = await agent.chat("北京今天天气怎么样？")
print(response)

await agent.close()

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

三、MCP 的生态与应用场景

MCP 协议的出现，为 AI 工具生态的发展提供了标准化的基础。以下是 MCP 的几个典型应用场景：

1. 数据库连接工具

传统方式下，AI 应用需要针对每种数据库（MySQL、PostgreSQL、MongoDB、Redis 等）编写专门的查询逻辑。使用 MCP，可以开发通用的数据库 MCP Server，AI 应用通过标准协议连接，无需关心底层数据库类型。

例如，开发一个"SQL Query Server"，提供 execute_query、list_tables、describe_table 等工具。任何支持 MCP 的 AI 应用都可以动态发现并调用这些工具，实现自然语言到 SQL 的转换和执行。

2. 文件系统访问

AI Agent 经常需要读取、写入、搜索文件。通过 MCP Server 提供文件系统访问能力，可以在保证安全的前提下，让 AI 应用操作本地或远程文件系统。

例如，"File System Server" 可以提供 read_file、write_file、list_directory、search_files 等工具。配合权限控制，可以限制 AI 应用只能访问特定目录，避免误删重要文件。

3. API 集成

企业内部的各种 API 服务（CRM、ERP、OA 等）可以通过 MCP Server 暴露给 AI 应用。这使得 AI Agent 能够自动调用企业系统，完成复杂的工作流。

例如，"CRM Server" 提供 create_contact、update_opportunity、generate_report 等工具。销售团队的 AI 助手可以通过这些工具自动更新客户信息、生成销售报告。

4. 开发工具集成

IDE 插件可以通过 MCP 调用各种开发工具（编译器、调试器、测试框架、代码格式化工具等），实现 AI 辅助编程。

例如，"Git Server" 提供 git_status、git_commit、git_push 等工具。AI 编程助手可以自动提交代码、创建 PR、解决合并冲突。

5. Web 浏览与搜索

MCP Server 可以提供 Web 浏览和搜索能力，使得 AI 应用能够获取实时信息。

例如，"Web Search Server" 提供 search、fetch_webpage、extract_content 等工具。AI 研究助手可以通过这些工具收集资料、总结网页内容。

# MCP 应用场景示例：企业 CRM 集成

# CRM MCP Server 实现

from mcp import Server, Tool
from mcp.server.stdio import stdio_server
from pydantic import BaseModel
from typing import Optional, List
import requests

class CreateContactInput(BaseModel):
first_name: str
last_name: str
email: str
phone: Optional[str] = None
company: Optional[str] = None

class UpdateOpportunityInput(BaseModel):
opportunity_id: str
stage: str # Prospecting, Qualification, Proposal, Negotiation, Closed Won, Closed Lost
amount: Optional[float] = None
close_date: Optional[str] = None

class CRMServer:
"""CRM 系统的 MCP Server 实现"""

def __init__(self, crm_api_url: str, api_key: str):
self.api_url = crm_api_url
self.api_key = api_key
self.server = Server("crm-server")
self._register_tools()

def _register_tools(self):
"""注册 CRM 工具"""

@self.server.tool(
name="create_contact",
description="在 CRM 系统中创建新联系人",
input_schema=CreateContactInput
)
async def create_contact(
first_name: str,
last_name: str,
email: str,
phone: Optional[str] = None,
company: Optional[str] = None
) -> dict:
"""创建联系人"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.api_url}/contacts",
json={
"first_name": first_name,
"last_name": last_name,
"email": email,
"phone": phone,
"company": company
},
headers={"Authorization": f"Bearer {self.api_key}"}
)

if response.status_code == 201:
return {"success": True, "contact_id": response.json()["id"]}
else:
return {"success": False, "error": response.text}

@self.server.tool(
name="update_opportunity",
description="更新 CRM 系统中的销售机会",
input_schema=UpdateOpportunityInput
)
async def update_opportunity(
opportunity_id: str,
stage: str,
amount: Optional[float] = None,
close_date: Optional[str] = None
) -> dict:
"""更新销售机会"""
async with httpx.AsyncClient() as client:
response = await client.patch(
f"{self.api_url}/opportunities/{opportunity_id}",
json={
"stage": stage,
"amount": amount,
"close_date": close_date
},
headers={"Authorization": f"Bearer {self.api_key}"}
)

if response.status_code == 200:
return {"success": True, "opportunity_id": opportunity_id}
else:
return {"success": False, "error": response.text}

@self.server.tool(
name="list_contacts",
description="列出 CRM 系统中的联系人"
)
async def list_contacts(limit: int = 10, company: Optional[str] = None) -> dict:
"""列出联系人"""
params = {"limit": limit}
if company:
params["company"] = company

async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.api_url}/contacts",
params=params,
headers={"Authorization": f"Bearer {self.api_key}"}
)

return {
"success": True,
"contacts": response.json()["data"]
}

@self.server.tool(
name="generate_sales_report",
description="生成销售报告"
)
async def generate_sales_report(
start_date: str,
end_date: str,
group_by: str = "sales_rep" # sales_rep, region, product
) -> dict:
"""生成销售报告"""
# 实现逻辑…
pass

async def run(self):
"""启动 Server"""
await stdio_server(self.server)

# MCP Client 示例：销售 AI 助手

class SalesAssistant:
"""销售 AI 助手，使用 MCP 调用 CRM 工具"""

def __init__(self, mcp_server_command: list):
self.agent = MCPEnabledAgent(mcp_server_command)

async def initialize(self):
"""初始化"""
await self.agent.initialize()

async def handle_request(self, user_input: str):
"""处理销售团队的请求"""

# 示例请求：
# – "帮我创建一个新联系人：张三，邮箱 zhangsan@example.com，公司 ABC Inc"
# – "把机会 OP-12345 的阶段更新为 Negotiation"
# – "生成本月的销售报告"

response = await self.agent.chat(user_input)
return response

# 使用示例
async def main():
import os

# 启动 CRM MCP Server
crm_server = CRMServer(
crm_api_url="https://api.crm-system.com",
api_key=os.environ["CRM_API_KEY"]
)

# 在单独进程中启动 Server
import subprocess
server_process = subprocess.Popen(["python", "crm_mcp_server.py"])

# 创建销售助手
assistant = SalesAssistant(mcp_server_command=["python", "crm_mcp_server.py"])
await assistant.initialize()

# 处理请求
response = await assistant.handle_request(
"帮我创建一个新联系人：李四，邮箱 lisi@example.com，公司 XYZ Corp"
)
print(response)

# 清理
server_process.terminate()
await assistant.agent.close()

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

四、MCP 的挑战与未来发展方向

尽管 MCP 协议为 AI 工具生态提供了标准化的基础，但在实际推广和应用中仍面临诸多挑战。

1. 性能开销

MCP 的客户端-服务器架构引入了额外的通信开销。每次工具调用都需要经过序列化和反序列化（JSON 格式），以及进程间通信（stdio 或 HTTP）。对于高频调用的场景（如批量数据处理），这种开销可能影响系统性能。

优化方向：

• 二进制序列化：使用 Protobuf、MessagePack 等二进制格式替代 JSON，减少序列化开销。
• 连接复用：保持长连接，避免频繁建立连接的开销。
• 批量调用：支持一次请求调用多个工具，减少往返次数。
• 本地缓存：缓存工具列表和 schema，减少不必要的查询。

2. 安全与沙箱

MCP Server 通常具有较高的系统权限（如文件系统访问、数据库连接）。恶意或错误的工具调用可能导致严重的安全问题。如何在保证功能性的同时，提供安全的执行环境，是 MCP 推广的关键。

优化方向：

• 沙箱隔离：使用容器（Docker）、虚拟机或 WebAssembly 沙箱隔离 MCP Server，限制其系统访问权限。
• 权限控制：实现细粒度的权限控制，用户可以配置每个 Server 的访问权限（只读、只写、禁止访问特定目录等）。
• 审计日志：记录所有工具调用请求和结果，便于安全审计和问题排查。
• 输入验证：对工具输入参数进行严格验证，防止注入攻击。

3. 工具质量与可靠性

MCP 协议本身不保证工具的质量和可靠性。一个设计不良的 MCP Server 可能导致工具调用失败、返回错误结果或性能低下。如何建立工具的质量标准和认证体系，是生态健康发展的基础。

优化方向：

• 工具注册表：建立类似 PyPI、npm 的中央工具注册表，提供工具搜索、版本管理、依赖解析等功能。
• 质量评分：基于用户评价、下载量、维护频率等指标，为工具提供质量评分。
• 兼容性测试：提供自动化测试工具，验证 MCP Server 是否符合协议规范。
• 文档规范：制定工具文档标准，要求提供清晰的功能描述、参数说明、使用示例和错误处理指南。

4. 跨平台兼容性

MCP 协议需要支持不同的操作系统（Linux、macOS、Windows）、编程语言（Python、TypeScript、Go、Rust）和 AI 框架（OpenAI、Anthropic、LangChain、Semantic Kernel）。如何确保跨平台兼容性，避免生态碎片化，是长期挑战。

优化方向：

• 多语言 SDK：提供官方 SDK，降低 MCP Server 和 Client 的开发门槛。
• 协议版本管理：制定清晰的版本管理策略，确保向后兼容性。
• 兼容性认证：提供兼容性认证程序，确保不同实现之间的互操作性。
• 社区驱动：建立开放的社区治理机制，鼓励各方参与协议演进。

5. 商业模式探索

MCP 工具生态的可持续发展需要清晰的商业模式。工具开发者如何变现？企业如何评估工具的 ROI？这些问题直接影响生态的活跃度。

可能的商业模式：

• 开源 + 商业支持：基础功能开源，高级功能或企业级支持收费（类似 Red Hat）。
• SaaS 订阅：MCP Server 作为云服务提供的，按调用次数或订阅收费。
• 工具市场：建立工具市场平台，开发者发布工具，用户购买使用，平台抽取佣金。
• 企业定制：为企业客户提供定制化的 MCP Server 开发服务。

# MCP 性能优化示例：批量工具调用

from mcp import Client
from typing import List, Dict
import asyncio

class OptimizedMCPClient:
"""优化性能的 MCP Client"""

def __init__(self, mcp_server_command: list):
self.client = stdio_client(mcp_server_command)
self.tool_cache = {} # 缓存工具列表

async def initialize(self):
"""初始化并缓存工具列表"""
await self.client.initialize()

# 获取并缓存工具列表
response = await self.client.list_tools()
for tool in response.tools:
self.tool_cache[tool.name] = tool

print(f"已缓存 {len(self.tool_cache)} 个工具")

async def batch_call_tools(self, calls: List[Dict]) -> List[Dict]:
"""
批量调用工具

Args:
calls: 工具调用请求列表，每个元素包含 name 和 input

Returns:
工具执行结果列表
"""
# 并发执行所有工具调用
tasks = []
for call in calls:
task = self.client.call_tool(call["name"], call["input"])
tasks.append(task)

results = await asyncio.gather(*tasks, return_exceptions=True)

# 处理结果
formatted_results = []
for i, result in enumerate(results):
if isinstance(result, Exception):
formatted_results.append({
"call": calls[i],
"error": str(result)
})
else:
formatted_results.append({
"call": calls[i],
"result": result
})

return formatted_results

async def cached_tool_lookup(self, tool_name: str) -> dict:
"""缓存的工具查找"""
if tool_name in self.tool_cache:
return self.tool_cache[tool_name]
else:
# 缓存未命中，重新查询
tools = await self.client.list_tools()
for tool in tools:
self.tool_cache[tool.name] = tool

return self.tool_cache.get(tool_name)

# 使用示例
async def main():
client = OptimizedMCPClient(["python", "mcp_server.py"])
await client.initialize()

# 批量调用工具
calls = [
{"name": "get_weather", "input": {"city": "北京"}},
{"name": "get_weather", "input": {"city": "上海"}},
{"name": "get_weather", "input": {"city": "广州"}},
{"name": "search_web", "input": {"query": "AI 最新进展"}}
]

results = await client.batch_call_tools(calls)
print(results)

await client.client.close()

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

结论

MCP（Model Context Protocol）协议为 AI 工具生态的标准化和互操作性提供了重要基础。通过统一工具定义格式、采用客户端-服务器架构、支持动态工具发现和上下文管理，MCP 有望解决当前 Function Calling 实现中的碎片化问题。

关键要点：

Function Calling 的下一步，不是更复杂的模型或更大的上下文窗口，而是更开放、更标准化、更安全的工具生态。MCP 协议是这个方向的重要探索，值得每个 AI 工程师关注。

参考资料

本文基于 MCP 协议规范和作者的工程实践。MCP 协议仍在快速演进，部分细节可能随时间变化。