引言

随着人工智能技术的快速发展，大语言模型（LLM）如ChatGPT、Claude等已经成为了改变我们工作和生活方式的强大工具。但你是否想过，如何让这些AI模型具备访问外部工具和数据的能力，从而解决更复杂的问题？今天，我将向大家介绍一项令人兴奋的技术——模型上下文协议（Model Context Protocol，简称MCP），并教你如何使用Python开发自己的MCP服务，为AI模型赋予更强大的能力。

什么是MCP？

基本概念

模型上下文协议（MCP）是一个开放标准，用于AI应用程序与大型语言模型之间的通信。它定义了一套标准接口，使应用程序能够向模型提供上下文信息，并允许模型调用应用程序暴露的工具。

简单来说，MCP就像是AI模型和外部世界之间的一座桥梁，让模型能够”看见”和”操作”外部的数据和功能。

为什么需要MCP？

想象一下，如果你正在与ChatGPT聊天，希望它能够：

  • 查询你的个人日历
  • 分析你的Excel数据
  • 控制你的智能家居设备
  • 从你的私有数据库中获取信息

这些功能都需要AI模型能够访问外部系统和数据，而MCP正是为解决这一需求而生的。

MCP的核心组件

MCP协议定义了三个核心原语，每一个都扮演着重要的角色：

原语 控制权 描述 示例应用
Prompts（提示） 用户控制 用户可以选择调用的交互式模板 快捷命令、应用菜单
Resources（资源） 应用程序控制 应用程序管理的上下文数据 文件内容、数据库记录
Tools（工具） 模型控制 AI模型可调用的函数 API请求、数据处理

这三种原语共同构成了MCP的基础，使模型能够以结构化的方式与外部世界交互。

准备工作：安装MCP Python SDK

在开始开发MCP服务之前，我们需要安装MCP的Python SDK。打开你的终端并运行：

1
pip install mcp

这个简单的命令将安装所有必要的依赖，让你能够立即开始开发。

创建你的第一个MCP服务

让我们从一个简单的例子开始，创建一个能够执行基本数学运算的MCP服务。

基本加法计算器

创建一个名为calculator.py的文件，并添加以下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from mcp.server.fastmcp import FastMCP

# 创建一个名为"简易计算器"的MCP服务器
mcp = FastMCP("简易计算器")

# 定义一个加法工具
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together"""
    return a + b

# 启动服务器
if __name__ == "__main__":
    mcp.run()

这段代码创建了一个简单的MCP服务，它提供了一个add工具，可以将两个数字相加。

运行你的MCP服务

现在，你可以在终端中运行这个服务：

1
2
3
python calculator.py
# 或者
mcp run calculator.py

一旦服务启动，它就可以被支持MCP的AI模型或应用程序使用了。

进阶：创建功能丰富的MCP服务

在了解了基础知识后，让我们创建一个更复杂的MCP服务，它包含了所有三种核心原语：提示、资源和工具。

多功能助手服务

创建一个名为assistant.py的文件，并添加以下代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from mcp.server.fastmcp import FastMCP
import datetime
import random

# 创建一个名为"多功能助手"的MCP服务器
mcp = FastMCP("多功能助手")

# 定义一个资源：当前时间
@mcp.resource("time://current")
def current_time() -> str:
    """Return the current time as a resource"""
    return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

# 定义一个工具：掷骰子
@mcp.tool()
def roll_dice(sides: int = 6) -> int:
    """Roll a dice with the specified number of sides"""
    return random.randint(1, sides)

# 定义一个工具：计算BMI
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
    """Calculate BMI given weight in kg and height in meters"""
    return weight_kg / (height_m * height_m)

# 定义一个提示：创建健康建议
@mcp.prompt()
def health_advice(user_name: str) -> str:
    """Generate a prompt for health advice"""
    return f"请为{user_name}提供一些基于科学的健康生活方式建议。考虑饮食、运动和睡眠方面的建议。"

# 启动服务器
if __name__ == "__main__":
    mcp.run()

这个多功能助手服务提供了:

  • 一个资源：获取当前时间
  • 两个工具：掷骰子和计算BMI
  • 一个提示模板：生成健康建议

高级功能：整合外部API

MCP服务的真正力量在于它能够将AI模型与外部系统和API集成。下面是一个示例，展示如何创建一个能够获取天气信息的MCP服务：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from mcp.server.fastmcp import FastMCP
import requests

# 创建一个名为"天气助手"的MCP服务器
mcp = FastMCP("天气助手")

API_KEY = "your_api_key_here"  # 替换为你的API密钥

@mcp.tool()
def get_weather(city: str) -> str:
    """Get current weather for a specified city"""
    url = f"https://api.weatherapi.com/v1/current.json?key={API_KEY}&q={city}"
    response = requests.get(url)
    if response.status_code == 200:
        data = response.json()
        temp_c = data["current"]["temp_c"]
        condition = data["current"]["condition"]["text"]
        humidity = data["current"]["humidity"]
        return f"City: {city}, Temperature: {temp_c}°C, Condition: {condition}, Humidity: {humidity}%"
    else:
        return f"Error fetching weather data: {response.status_code}"

# 启动服务器
if __name__ == "__main__":
    mcp.run()

这个服务通过调用外部的天气API，为模型提供了获取实时天气信息的能力。

将MCP服务与现有应用集成

如果你已经有一个现有的Web应用程序，你可以将MCP服务集成到其中。以下是如何将MCP服务与Starlette（一个轻量级的ASGI框架）集成的示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
from starlette.applications import Starlette
from starlette.routing import Mount
from mcp.server.fastmcp import FastMCP

# 创建MCP服务
mcp = FastMCP("集成服务")

@mcp.tool()
def greet(name: str) -> str:
    """Generate a greeting message"""
    return f"Hello, {name}! Welcome to our integrated service."

# 将MCP服务挂载到Starlette应用
app = Starlette(
    routes=[
        Mount('/mcp', app=mcp.sse_app()),
    ]
)

# 启动服务器
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

通过这种方式，你可以在不改变现有应用架构的情况下，为它添加AI能力。

高级概念：状态管理

有时，你需要在不同的请求之间保持状态。下面是一个简单的待办事项管理器，它展示了如何在MCP服务中管理状态：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from mcp.server.fastmcp import FastMCP

# 创建一个名为"待办事项管理器"的MCP服务器
mcp = FastMCP("待办事项管理器")

# 使用一个简单的列表来存储待办事项
todos = []

@mcp.tool()
def add_todo(task: str) -> str:
    """Add a new task to the todo list"""
    todos.append(task)
    return f"Task added: {task}"

@mcp.tool()
def list_todos() -> str:
    """List all tasks in the todo list"""
    if not todos:
        return "No tasks in the todo list."
    return "\n".join([f"{i+1}. {task}" for i, task in enumerate(todos)])

@mcp.tool()
def complete_todo(task_number: int) -> str:
    """Mark a task as completed and remove it from the list"""
    if not todos or task_number <= 0 or task_number > len(todos):
        return "Invalid task number."
    task = todos.pop(task_number - 1)
    return f"Completed task: {task}"

# 启动服务器
if __name__ == "__main__":
    mcp.run()

这个示例展示了如何使用全局变量来保存状态。在实际应用中，你可能需要使用数据库或其他持久化存储方案。

客户端与MCP服务的交互

MCP不仅定义了服务器端的接口，还定义了客户端如何与这些服务交互。下面是一个简单的客户端示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# 创建服务器参数
server_params = StdioServerParameters(
    command="python",
    args=["calculator.py"],
)

async def run():
    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(f"Available tools: {tools}")
            
            # 调用add工具
            result = await session.call_tool("add", arguments={"a": 5, "b": 3})
            print(f"5 + 3 = {result}")

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

这个客户端脚本连接到我们之前创建的计算器服务，列出可用的工具，然后调用add工具计算5+3的结果。

cursor中配置自己开发的MCP

在mcp.json中添加mcp server

1
2
3
4
5
6
7
8
9
10
{
    "mcpServers": {
        "server": {
            "command": "python",
            "args": [
                "/User/admin/workspace/mcp-server/main.py"
            ]
        }
    }
}

customize mcp

mcp 示例

实际应用场景

MCP服务的应用场景非常广泛，以下是一些可能的使用案例：

  1. 个人助手：创建一个能够访问你的日历、邮件和文档的个人助手

  2. 数据分析工具：开发一个能够处理和分析数据集的服务，让AI模型可以直接操作数据

  3. 内容管理系统：构建一个服务，允许AI模型创建、编辑和发布内容到你的网站或博客

  4. 企业知识库：开发一个能够安全访问企业内部文档和知识库的服务

  5. 智能家居控制：创建一个服务，让AI能够控制你的智能家居设备

调试技巧

在开发MCP服务时，调试是一个重要的环节。以下是一些有用的调试技巧：

  1. 启用详细日志：在创建FastMCP实例时，可以启用详细日志：
1
mcp = FastMCP("调试服务", verbose=True)
  1. 添加自定义日志：在工具和资源处理函数中，添加打印语句：
1
2
3
4
5
@mcp.tool()
def my_tool(param: str) -> str:
    print(f"my_tool called with param: {param}")
    # 处理逻辑
    return result
  1. 使用调试客户端：创建一个简单的客户端脚本，专门用于测试你的服务

总结

MCP为开发者提供了一种强大的方式，让AI模型能够与外部世界交互，从而扩展其能力范围。通过使用Python和MCP SDK，即使是初学者也能轻松地开发出功能丰富的MCP服务，为AI模型赋予更强大的能力。

无论你是想创建个人助手、企业工具，还是创新的AI应用，MCP都为你提供了所需的基础设施。随着对MCP的深入学习和实践，你将能够构建出更加复杂和强大的系统，充分发挥AI技术的潜力。

学习资源

开始你的MCP开发之旅吧，相信不久的将来，你将能够创建出令人惊叹的AI驱动应用！

本文由 AI 辅助生成，如有错误或建议，欢迎指出。