MCP协议深度解析：AI Agent如何通过标准接口调用外部工具

# MCP协议深度解析：AI Agent如何通过标准接口调用外部工具

## 摘要

Model Context Protocol（MCP）是Anthropic提出的开放标准，旨在解决AI Agent与外部工具之间的接口碎片化问题。本文深入解析MCP协议的架构设计、传输机制、资源管理，并实战构建MCP Server和Client。

## 一、为什么需要MCP？

### 1.1 现状：工具接入的碎片化

当前AI Agent调用外部工具的方式五花八门：

```

OpenAI → Function Calling（自定义JSON Schema）

Anthropic → Tool Use（另一种JSON Schema）

LangChain → Tool抽象层（Python函数包装）

自定义Agent → 硬编码API调用

```

每接入一个新工具，都需要：

- 适配不同LLM的函数调用格式

- 处理不同的认证方式

- 管理不同工具的错误处理逻辑

- 重复编写连接代码

### 1.2 MCP的解决方案

MCP的核心设计理念：**像USB-C统一接口一样，统一AI Agent与工具之间的连接标准**。

```

┌──────────────┐ MCP协议 ┌──────────────┐

│ AI Agent │ ←─────────────→ │ MCP Server │

│ (Client) │ 标准接口 │ (工具提供方) │

└──────────────┘ └──────┬───────┘

│

┌──────┴───────┐

│ 实际工具/资源 │

│ DB/API/文件等 │

└──────────────┘

```

## 二、MCP协议架构

### 2.1 核心概念

MCP定义了三个核心原语：

| 原语 | 方向 | 用途 |

|-----|------|------|

| **Tools** | Client→Server | Agent调用工具执行操作 |

| **Resources** | Server→Client | Server暴露可读取的数据/内容 |

| **Prompts** | Server→Client | Server提供预定义的Prompt模板 |

### 2.2 通信模型

MCP采用JSON-RPC 2.0协议，支持两种传输方式：

**Stdio传输**（本地进程间通信）：

```

Agent进程 ← stdin/stdout → MCP Server进程

```

**SSE传输**（远程HTTP通信）：

```

Agent ← HTTP POST → MCP Server

Agent ← SSE GET ← MCP Server（服务端推送）

```

### 2.3 消息类型

```json

// 请求

{

"jsonrpc": "2.0",

"id": 1,

"method": "tools/call",

"params": {

"name": "query_database",

"arguments": {

"sql": "SELECT * FROM articles WHERE article_type = 1 LIMIT 5"

}

}

}

// 响应

{

"jsonrpc": "2.0",

"id": 1,

"result": {

"content": [

{

"type": "text",

"text": "[{\"id\": 1, \"title\": \"...\", ...}]"

}

]

}

}

```

## 三、实战：构建MCP Server

### 3.1 安装MCP SDK

```bash

pip install mcp

```

### 3.2 创建数据库查询MCP Server

```python

# db_mcp_server.py

from mcp.server import Server

from mcp.server.stdio import stdio_server

from mcp.types import Tool, TextContent

import mysql.connector

app = Server("db-query-server")

DB_CONFIG = {

'host': '59.110.223.195',

'port': 7889,

'user': 'root',

'password': 'Renjnet@1',

'database': 'info',

'charset': 'utf8mb4'

}

@app.list_tools()

async def list_tools() -> list[Tool]:

return [

Tool(

name="query_articles",

description="查询info数据库中的文章，支持按分类、关键词筛选",

inputSchema={

"type": "object",

"properties": {

"article_type": {

"type": "integer",

"description": "文章类型：1=AI, 2=运维, 3=开发, 5=短视频和直播, 6=其他科技",

"enum": [1, 2, 3, 5, 6]

},

"keyword": {

"type": "string",

"description": "搜索关键词"

},

"limit": {

"type": "integer",

"description": "返回条数，默认5",

"default": 5

}

},

"required": []

}

),

Tool(

name="get_article_detail",

description="根据文章GUID获取完整文章内容",

inputSchema={

"type": "object",

"properties": {

"guid": {

"type": "string",

"description": "文章唯一标识GUID"

}

},

"required": ["guid"]

}

)

]

@app.call_tool()

async def call_tool(name: str, arguments: dict) -> list[TextContent]:

conn = mysql.connector.connect(**DB_CONFIG)

cursor = conn.cursor(dictionary=True)

try:

if name == "query_articles":

sql = "SELECT guid, title, excerpt, article_type, pub_time FROM articles WHERE 1=1"

params = []

if "article_type" in arguments:

sql += " AND article_type = %s"

params.append(arguments["article_type"])

if "keyword" in arguments:

sql += " AND (title LIKE %s OR keywords LIKE %s)"

kw = f"%{arguments['keyword']}%"

params.extend([kw, kw])

sql += " ORDER BY pub_time DESC LIMIT %s"

params.append(arguments.get("limit", 5))

cursor.execute(sql, params)

results = cursor.fetchall()

return [TextContent(type="text", text=str(results))]

elif name == "get_article_detail":

cursor.execute(

"SELECT * FROM articles WHERE guid = %s",

(arguments["guid"],)

)

result = cursor.fetchone()

return [TextContent(type="text", text=str(result))]

finally:

cursor.close()

conn.close()

async def main():

async with stdio_server() as (read_stream, write_stream):

await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":

import asyncio

asyncio.run(main())

```

### 3.3 MCP Server配置

在Claude Desktop或支持MCP的客户端中配置：

```json

{

"mcpServers": {

"db-query": {

"command": "python",

"args": ["D:/work/ai/codes/mcp_servers/db_mcp_server.py"]

}

}

}

```

## 四、实战：构建MCP Client

```python

# mcp_client.py

from mcp import ClientSession, StdioServerParameters

from mcp.client.stdio import stdio_client

server_params = StdioServerParameters(

command="python",

args=["db_mcp_server.py"]

)

async def query_via_mcp():

async with stdio_client(server_params) as (read, write):

async with ClientSession(read, write) as session:

await session.initialize()

# 列出可用工具

tools = await session.list_tools()

print("可用工具:", [t.name for t in tools.tools])

# 调用工具

result = await session.call_tool(

"query_articles",

arguments={"article_type": 1, "limit": 3}

)

print("查询结果:", result.content[0].text)

import asyncio

asyncio.run(query_via_mcp())

```

## 五、MCP vs Function Calling

| 维度 | Function Calling | MCP |

|-----|-----------------|-----|

| 接口标准 | 厂商私有（OpenAI/Anthropic各自定义） | 开放标准 |

| 工具发现 | 需硬编码在Prompt中 | 动态发现（list_tools） |

| 运行方式 | 同步，嵌入LLM调用 | 独立进程/服务 |

| 可复用性 | 低（每个Agent重新定义） | 高（一次开发，多Agent复用） |

| 安全控制 | 依赖LLM方 | Server端独立控制权限 |

| 生态成熟度 | 高（广泛支持） | 快速增长中 |

## 六、MCP生态现状（2026年）

| 类别 | 示例 |

|------|------|

| 数据库 | PostgreSQL、MySQL、SQLite、MongoDB |

| 文件系统 | 本地文件、S3、GCS |

| 开发工具 | GitHub、GitLab、Jira |

| 通信 | Slack、Email、企业微信 |

| 搜索 | Brave Search、Tavily |

| 创作工具 | Figma、Notion |

## 总结

MCP通过定义AI Agent与外部工具之间的标准通信协议，解决了工具接入的碎片化问题。虽然生态仍在发展中，但其"一次开发、多Agent复用"的理念代表了AI工具集成的未来方向。对于企业内部AI应用，构建MCP Server是提升工具可复用性和安全性的有效途径。

---

*本文由北科信息日采集系统自动生成，发布日期：2026-05-05*