AI辅助创作 | 专栏《2026 AI编程效率革命》第07篇

前言

MCP（Model Context Protocol）是Anthropic在2024年底推出的开放协议，旨在标准化AI模型与外部工具、数据源的交互方式。到2026年，MCP已经成为AI Agent开发的事实标准协议。本文将从零开始，带你理解MCP协议的核心概念，并通过实战代码搭建一个完整的MCP服务。

---

一、MCP协议是什么？

1.1 为什么需要MCP？

在MCP出现之前，AI工具集成面临以下问题：

• 每个AI应用都要为每个工具写单独的集成代码
• 工具接口没有统一标准，重复造轮子
• 模型切换时工具链需要重写

MCP的核心理念：像USB-C统一接口一样，统一AI与工具的连接方式。

1.2 MCP架构

┌─────────────┐     MCP协议     ┌─────────────┐
│   AI应用     │ ◄──────────────► │  MCP Server  │
│  (Host)      │                 │  (工具提供方)  │
│  如: Cursor  │                 │              │
└─────────────┘                 └──────┬───────┘
                                       │
                                ┌──────┴───────┐
                                │  外部资源      │
                                │  数据库/API    │
                                │  文件系统等    │
                                └──────────────┘

MCP定义了三种核心能力：

能力	说明	示例
Tools	可调用的函数	查询数据库、发送邮件
Resources	可读取的数据	文件内容、数据库记录
Prompts	预定义提示词	代码审查模板

---

二、MCP开发环境搭建

2.1 安装MCP SDK

# 安装MCP Python SDK
uv add mcp

# 安装官方示例依赖
uv add httpx pydantic

2.2 项目结构

mcp-demo/
├── pyproject.toml
├── .env
├── server/
│   ├── __init__.py
│   ├── weather_server.py    # 天气查询MCP服务
│   ├── file_server.py       # 文件管理MCP服务
│   └── db_server.py         # 数据库查询MCP服务
├── client/
│   ├── __init__.py
│   └── mcp_client.py        # MCP客户端
└── tests/
    └── test_server.py

---

三、实战：创建你的第一个MCP Server

3.1 天气查询MCP服务

# server/weather_server.py
import httpx
from mcp.server import Server
from mcp.types import Tool, TextContent

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

@server.list_tools()
async def list_tools() -> list[Tool]:
    """声明服务提供的工具列表"""
    return [
        Tool(
            name="get_weather",
            description="获取指定城市的当前天气信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如'北京'、'上海'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，默认摄氏度",
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="get_forecast",
            description="获取未来3天天气预报",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    }
                },
                "required": ["city"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """处理工具调用请求"""
    if name == "get_weather":
        return await get_weather(arguments["city"], arguments.get("unit", "celsius"))
    elif name == "get_forecast":
        return await get_forecast(arguments["city"])
    else:
        raise ValueError(f"未知工具: {name}")

async def get_weather(city: str, unit: str = "celsius") -> list[TextContent]:
    """查询天气（使用模拟数据）"""
    # 实际项目中调用天气API
    weather_data = {
        "北京": {"temp": 22, "humidity": 45, "condition": "晴"},
        "上海": {"temp": 25, "humidity": 72, "condition": "多云"},
        "深圳": {"temp": 30, "humidity": 80, "condition": "阵雨"},
    }
    
    data = weather_data.get(city, {"temp": 20, "humidity": 50, "condition": "未知"})
    
    if unit == "fahrenheit":
        data["temp"] = data["temp"] * 9/5 + 32
    
    result = (
        f"📍 {city} 当前天气\n"
        f"🌡️ 温度: {data['temp']}{'°C' if unit == 'celsius' else '°F'}\n"
        f"💧 湿度: {data['humidity']}%\n"
        f"🌤️ 状况: {data['condition']}"
    )
    
    return [TextContent(type="text", text=result)]

async def get_forecast(city: str) -> list[TextContent]:
    """获取天气预报"""
    forecast = (
        f"📍 {city} 未来3天预报\n"
        f"明天: 晴，18-26°C\n"
        f"后天: 多云，17-24°C\n"
        f"大后天: 小雨，15-22°C"
    )
    return [TextContent(type="text", text=forecast)]

# 启动服务
if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server
    
    async def main():
        async with stdio_server() as (read_stream, write_stream):
            await server.run(read_stream, write_stream)
    
    asyncio.run(main())

3.2 文件管理MCP服务

# server/file_server.py
import os
import json
from pathlib import Path
from mcp.server import Server
from mcp.types import Tool, TextContent

server = Server("file-manager")
WORKSPACE = Path("./workspace")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="list_files",
            description="列出指定目录下的文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "目录路径（相对于workspace）"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="read_file",
            description="读取文件内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "文件路径"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="write_file",
            description="写入文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "content": {"type": "string"}
                },
                "required": ["path", "content"]
            }
        ),
        Tool(
            name="search_files",
            description="按名称搜索文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "pattern": {
                        "type": "string",
                        "description": "搜索关键词"
                    }
                },
                "required": ["pattern"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "list_files":
        target = WORKSPACE / arguments["path"]
        files = [str(f.relative_to(WORKSPACE)) for f in target.rglob("*") if f.is_file()]
        return [TextContent(type="text", text=json.dumps(files[:50], indent=2))]
    
    elif name == "read_file":
        target = WORKSPACE / arguments["path"]
        content = target.read_text(encoding="utf-8")
        return [TextContent(type="text", text=content)]
    
    elif name == "write_file":
        target = WORKSPACE / arguments["path"]
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(arguments["content"], encoding="utf-8")
        return [TextContent(type="text", text=f"已写入: {arguments['path']}")]
    
    elif name == "search_files":
        results = [
            str(f.relative_to(WORKSPACE))
            for f in WORKSPACE.rglob(f"*{arguments['pattern']}*")
            if f.is_file()
        ]
        return [TextContent(type="text", text=json.dumps(results[:20], indent=2))]
    
    raise ValueError(f"未知工具: {name}")

if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server
    
    async def main():
        async with stdio_server() as (read_stream, write_stream):
            await server.run(read_stream, write_stream)
    
    asyncio.run(main())

---

四、MCP客户端开发

4.1 连接MCP Server

# client/mcp_client.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MCPClient:
    def __init__(self):
        self.session: ClientSession | None = None
        self.available_tools: list = []

    async def connect(self, server_command: list[str]):
        """连接到MCP Server"""
        server_params = StdioServerParameters(
            command=server_command[0],
            args=server_command[1:],
        )
        
        self.stdio_context = stdio_client(server_params)
        read_stream, write_stream = await self.stdio_context.__aenter__()
        
        self.session = ClientSession(read_stream, write_stream)
        await self.session.__aenter__()
        await self.session.initialize()
        
        # 获取可用工具列表
        result = await self.session.list_tools()
        self.available_tools = result.tools
        print(f"已连接，可用工具: {[t.name for t in self.available_tools]}")

    async def call_tool(self, tool_name: str, arguments: dict):
        """调用工具"""
        if not self.session:
            raise RuntimeError("未连接到MCP Server")
        result = await self.session.call_tool(tool_name, arguments)
        return result.content

    async def disconnect(self):
        """断开连接"""
        if self.session:
            await self.session.__aexit__(None, None, None)
            await self.stdio_context.__aexit__(None, None, None)

async def main():
    client = MCPClient()
    
    # 连接天气服务
    await client.connect(["python", "server/weather_server.py"])
    
    # 调用工具
    result = await client.call_tool("get_weather", {"city": "北京"})
    print(result[0].text)
    
    result = await client.call_tool("get_forecast", {"city": "上海"})
    print(result[0].text)
    
    await client.disconnect()

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

---

五、与AI模型集成

5.1 将MCP工具接入LangChain Agent

# integration/langchain_mcp_agent.py
import asyncio
import json
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from client.mcp_client import MCPClient

# 创建MCP客户端
mcp_client = MCPClient()

@tool
async def query_weather(city: str) -> str:
    """查询指定城市的天气信息"""
    result = await mcp_client.call_tool("get_weather", {"city": city})
    return result[0].text

@tool
async def get_forecast(city: str) -> str:
    """获取城市未来3天天气预报"""
    result = await mcp_client.call_tool("get_forecast", {"city": city})
    return result[0].text

async def run_agent():
    # 连接MCP服务
    await mcp_client.connect(["python", "server/weather_server.py"])
    
    # 创建LangChain Agent
    llm = ChatOpenAI(model="gpt-5.5", temperature=0)
    tools = [query_weather, get_forecast]
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个天气助手，可以查询天气和预报。用中文回复。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
    
    agent = create_tool_calling_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    
    # 测试
    result = await executor.ainvoke({"input": "北京和上海今天天气怎么样？明天呢？"})
    print(result["output"])
    
    await mcp_client.disconnect()

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

---

六、在Cursor中使用MCP

Cursor原生支持MCP协议，配置方法：

// .cursor/mcp.json
{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["server/weather_server.py"]
    },
    "file-manager": {
      "command": "python",
      "args": ["server/file_server.py"]
    }
  }
}

配置完成后，在Cursor的Agent模式中就可以直接调用这些MCP工具了。

---

七、MCP生态与社区资源

2026年MCP生态已经相当成熟：

• MCP Hub：官方工具市场，数百个现成MCP Server
• 常用MCP Server：数据库(PostgreSQL/MySQL)、GitHub、Slack、Google Drive等
• 多语言SDK：Python、TypeScript、Java、Go

---

总结

MCP协议通过标准化AI与工具的交互方式，解决了重复集成的问题。本文从协议原理讲到实战代码，覆盖了：

1. MCP的核心概念（Tools/Resources/Prompts）
2. MCP Server开发实战
3. MCP客户端开发
4. 与LangChain Agent集成
5. 在Cursor中使用MCP

下一篇我们将深入多Agent协作系统的搭建，利用MCP协议构建更复杂的AI应用。

---

免责声明：本文为AI辅助创作内容，代码示例仅供学习参考。MCP协议规范可能随版本更新而变化，请以官方文档为准。实际项目中请注意API Key安全和数据隐私保护。

专栏：《2026 AI编程效率革命》| 第07篇
发布日期：2026-05-03