MCP（模型上下文协议）简单案例

MCP（模型上下文协议）的标准化接口

MCP（Model Context Protocol，模型上下文协议）是由Anthropic开发的一种开放协议，旨在标准化大型语言模型（LLM）与外部数据源和工具的交互方式1。它就像AI应用程序的USB-C接口，为AI模型提供了一种标准化的方式来连接不同的数据源和工具1。

MCP的核心功能

MCP协议包含以下几个核心功能：

1. Resources（资源）：允许LLM访问外部数据源，如文件、数据库或API1。

2. Tools（工具）：提供LLM可以调用的函数或操作，扩展模型的能力1。

3. Prompts（提示词）：预定义的提示模板，可用于指导模型生成特定类型的输出1。

4. Sampling（采样）：管理模型生成过程中的参数调整1。

5. Roots（根目录）：定义资源访问的起始点1。

6. Transports（传输层）：支持多种协议实现，包括stdio（标准输入/输出）和SSE（服务器发送事件）1。

MCP的架构

MCP遵循客户端-服务器架构，主要包含以下组件：

1. MCP主机（MCP Hosts）：发起请求的LLM应用程序，如聊天机器人、AI工具等14。

2. MCP客户端（MCP Clients）：在主机程序内部，与MCP服务器保持连接14。

3. MCP服务器（MCP Servers）：提供数据、工具和服务的外部系统14。

如何通过MCP暴露AI代理服务

1. 开发MCP服务器

要暴露您的AI代理服务，首先需要开发一个MCP服务器。以Python为例，您可以使用官方提供的SDK进行开发：

import httpx
from mcp.server import FastMCP# 初始化FastMCP服务器
app = FastMCP('your-service-name')# 使用@app.tool()装饰器定义工具
@app.tool()
async def your_ai_agent_function(parameter1: str, parameter2: int) -> str:"""您的AI代理服务描述Args:parameter1: 参数1的描述parameter2: 参数2的描述Returns:返回值描述"""# 实现您的AI代理逻辑return "结果"# 运行服务器
if __name__ == "__main__":app.run(transport='stdio')  # 或使用'sse'作为传输层

2. 选择传输协议

MCP支持两种主要的传输协议：

• stdio（标准输入/输出）：更为常用，适合本地运行1。
• SSE（服务器发送事件）：适用于远程服务1。

3. 注册和发布MCP服务器

开发完成后，您可以通过以下方式使您的MCP服务可被其他应用访问：

1. 本地安装：通过mcp install your_script.py命令安装到支持MCP的客户端应用（如Claude Desktop）中1。

2. 发布到npm：如果是JavaScript/TypeScript实现，可以将您的MCP服务器发布到npm供其他开发者使用21。

3. 使用远程代理：通过MCP-Connect等工具，将基于本地stdio的MCP服务器通过HTTP/HTTPS请求暴露给远程客户端7。

4. 与客户端集成

支持MCP的客户端（如Claude Desktop、Cursor、VSCode的Cline插件等）可以通过配置文件连接到您的MCP服务：

{"mcpServers": {"your-service-name": {"command": "运行命令","args": ["参数1", "参数2"],"env": {}}}
}

5. 集成到现有框架

MCP可以与多种现有框架集成：

• LangChain：通过langchain-mcp-adapters可以将MCP工具无缝集成到LangChain工作流中1。

• LlamaIndex：可以将MCP工具转换为可直接使用的LlamaIndex工具23。

实际应用案例

1. 图像生成：可以使用huggingface上的模型，如FLUX.1-schnell，创建图像生成MCP服务器1。

2. 网络搜索：开发网络搜索工具，让AI能够获取最新信息1。

3. 数据库交互：创建与MySQL、PostgreSQL、SQLite等数据库交互的MCP服务器7。

4. API集成：如APISIX-MCP，通过自然语言交互完成APISIX网关资源的查询与管理21。

5. 物联网控制：通过将MCP协议桥接到MQTT，实现通过自然语言指令控制智能家居和IoT设备20。

---

MCP实际开发示例与调试

1. 基于网络搜索的MCP服务器开发详细步骤

以下是开发一个网络搜索MCP服务器的完整过程：

import httpx
from mcp.server import FastMCP# 初始化 FastMCP 服务器
app = FastMCP('web-search')@app.tool()
async def web_search(query: str) -> str:"""搜索互联网内容Args:query: 要搜索内容Returns:搜索结果的总结"""async with httpx.AsyncClient() as client:response = await client.post('https://open.bigmodel.cn/api/paas/v4/tools',headers={'Authorization': '您的API KEY'},json={'tool': 'web-search-pro','messages': [{'role': 'user', 'content': query}],'stream': False})res_data = []for choice in response.json()['choices']:for message in choice['message']['tool_calls']:search_results = message.get('search_result')if not search_results:continuefor result in search_results:res_data.append(result['content'])return '\n\n\n'.join(res_data)if __name__ == "__main__":app.run(transport='stdio')

2. 使用MCP Inspector进行调试

MCP提供了可视化工具MCP Inspector来调试你的服务器：

# 通过npx运行Inspector
npx -y @modelcontextprotocol/inspector uv run web_search.py# 或通过mcp dev命令运行
mcp dev web_search.py

运行后，在浏览器打开提供的地址（通常是http://127.0.0.1:5173），点击Connect按钮连接你的服务器，然后在Tools标签中可以测试你的工具1。

3. 实现Sampling功能

Sampling功能允许在工具执行前后进行操作，例如用户确认删除文件的场景：

# 服务端
from mcp.server import FastMCP
from mcp.types import SamplingMessage, TextContentapp = FastMCP('file_server')@app.tool()
async def delete_file(file_path: str):# 创建SamplingMessage触发sampling callback函数result = await app.get_context().session.create_message(messages=[SamplingMessage(role='user', content=TextContent(type='text', text=f'是否要删除文件: {file_path} (Y)'))],max_tokens=100)# 根据返回值执行操作if result.content.text == 'Y':return f'文件 {file_path} 已被删除！！'

在客户端中设置相应的sampling_callback函数处理用户交互1。

4. 开发Prompt和Resource功能

from mcp.server import FastMCPapp = FastMCP('prompt_and_resources')# 定义Prompt模板
@app.prompt('翻译专家')
async def translate_expert(target_language: str = 'Chinese',
) -> str:return f'你是一个翻译专家，擅长将任何语言翻译成{target_language}。请翻译以下内容：'# 定义静态资源
@app.resource('echo://static')
async def echo_resource():return 'Echo!'# 定义带参数的资源模板
@app.resource('greeting://{name}')
async def get_greeting(name):return f'Hello, {name}!'

适配现有大模型

1. 使用DeepSeek等非Claude模型

创建一个MCPClient类来调用你的MCP服务，使非Claude模型也能利用MCP功能：

import json
import asyncio
import os
from typing import Optional
from contextlib import AsyncExitStackfrom openai import OpenAI
from dotenv import load_dotenvfrom mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_clientload_dotenv()class MCPClient:def __init__(self):self.session: Optional[ClientSession] = Noneself.exit_stack = AsyncExitStack()self.client = OpenAI()  # 使用DeepSeek或其他OpenAI兼容接口async def connect_to_server(self):# 连接到MCP服务器server_params = StdioServerParameters(command='uv',args=['run', 'your_mcp_server.py'])# 建立连接stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))stdio, write = stdio_transportself.session = await self.exit_stack.enter_async_context(ClientSession(stdio, write))await self.session.initialize()

在process_query方法中，你可以利用function calling机制让DeepSeek等模型调用MCP服务1。

大型项目中的MCP应用案例

1. APISIX-MCP：API管理的智能化

APISIX-MCP通过自然语言交互完成API网关资源的管理，支持以下操作：

• 通用操作：检索和删除资源
• API资源操作：管理路由、服务、上游、SSL证书等
• 插件操作：管理插件配置、全局规则等
• 安全配置：管理密钥、消费者、凭证等6

配置示例：

{"mcpServers": {"apisix-mcp": {"command": "npx","args": ["-y", "apisix-mcp"],"env": {"APISIX_SERVER_HOST": "your-apisix-server-host","APISIX_ADMIN_API_PORT": "your-apisix-admin-api-port"}}}
}

这种AI+MCP的组合显著降低了API管理的认知成本，提升了运维效率6。

2. mcp2mqtt：连接智能家居设备

该项目将MCP协议桥接到MQTT，实现通过自然语言指令控制智能家居、IoT设备和机器人，特别适合物联网开发者学习和应用8。