1. 项目起航：为什么我们需要亲手打造一个MCP智能体？

大家好，我是老张，一个在AI和智能硬件领域摸爬滚打了十多年的开发者。最近几年，大语言模型（LLM）的发展速度简直让人眼花缭乱，从最初的聊天对话，到现在能写代码、做分析、甚至帮你规划旅行。但不知道你有没有遇到过这样的困境：当你兴冲冲地问一个AI助手“帮我查一下我GitHub上最新的issue，然后总结一下发到我的Slack频道”，它大概率会礼貌地告诉你“我无法访问外部数据”。那一刻的无力感，就像你有一把万能钥匙，却打不开自家仓库的门。

这就是传统AI应用面临的“数据孤岛”问题。模型本身很强大，但它被关在了一个信息真空的房间里。过去，我们想让它连接外部世界，比如数据库、API或者本地文件，就得为每一个数据源、每一个工具编写独立的适配代码。这个过程不仅繁琐，而且一旦你想换个模型或者增加新功能，整个架构可能就得推倒重来，维护成本高得吓人。

直到我遇到了 MCP（Model Context Protocol，模型上下文协议）。你可以把它理解为AI世界的“USB-C”接口。以前，你的手机、电脑、平板各有各的充电口，出门得带一堆线。现在，一个USB-C接口全搞定。MCP干的就是这事儿，它为LLM和外部工具、数据源之间定义了一套标准化的通信协议。这意味着，你只需要按照这个协议“造”好一个“插座”（MCP服务器），任何支持这个协议的“电器”（LLM客户端）都能即插即用。

所以，今天我想带你从零开始，亲手构建一个属于你自己的MCP智能体。我们的目标很明确：将一个现有的LLM（比如开源的Llama 3或者DeepSeek-R1）变成一个能真正干活儿的自动化助手。它不仅能理解你的复杂指令，还能自主调用GitHub API去查询你的代码仓库，能连接本地数据库获取业务数据，甚至能调用天气API、发送邮件，把多步骤的任务一口气搞定。这不再是纸上谈兵的概念，而是一份手把手、可落地的实战指南。准备好了吗？我们这就开始。

2. 磨刀不误砍柴工：开发环境与核心概念梳理

在动手写代码之前，我们得先把“战场”布置好，并且彻底理解我们要用的“武器”。这一步看似枯燥，却能让你在后续开发中避开无数个坑。我刚开始玩MCP的时候，就是没重视环境配置，结果在依赖问题上折腾了大半天。

2.1 搭建你的开发环境

我们的技术栈会以Python为主，因为它生态丰富，而且MCP的官方SDK对Python支持非常友好。当然，你用Node.js或者Go也完全没问题，协议是通用的。

首先，我强烈建议你使用 Python 3.10或更高版本，并且创建一个独立的虚拟环境。这能避免和你系统里其他项目的依赖打架。

# 1. 创建项目目录并进入
mkdir my-mcp-agent && cd my-mcp-agent

# 2. 创建虚拟环境（以venv为例）
python -m venv venv

# 3. 激活虚拟环境
# 在 macOS/Linux 上：
source venv/bin/activate
# 在 Windows 上：
venv\Scripts\activate

# 4. 安装核心依赖
pip install mcp[cli] httpx sqlalchemy python-dotenv

简单解释一下这几个包：

• mcp[cli]：这是Anthropic官方维护的MCP Python SDK，[cli]额外安装了命令行工具，方便我们测试和调试服务器。
• httpx：一个现代、异步的HTTP客户端，我们用来调用外部API（比如GitHub API、天气API）。
• sqlalchemy：一个强大的ORM（对象关系映射）工具，让我们能用Python对象的方式来操作数据库，非常方便。
• python-dotenv：用来管理环境变量，比如你的GitHub个人访问令牌、数据库密码等敏感信息，千万不要硬编码在代码里！

提示：记得把 venv 目录添加到你的 .gitignore 文件里，不要把它提交到版本控制。

2.2 深入理解MCP的核心组件

安装好环境，我们来聊聊MCP里最重要的三个概念：资源（Resources）、工具（Tools） 和 提示（Prompts）。理解它们，就等于理解了MCP智能体的骨架。

资源（Resources） 你可以把它想象成智能体可以“看”到的静态数据文件。比如一个本地的 README.md 文件，一个数据库里的某张表视图，或者一个远程的API文档页面。资源的特点是“只读”，智能体可以读取它们的内容作为上下文，但不能直接修改。在我们的项目里，我计划把项目根目录下的 docs 文件夹作为资源暴露给智能体，这样它就能参考我们的项目文档来回答问题了。

工具（Tools） 这是智能体的“手”和“脚”，是它可以主动调用的函数。这才是让智能体活起来的关键！工具可以做任何事情：查询数据库、调用第三方API、读写文件（需谨慎）、发送通知等等。每个工具都有明确的输入参数（由JSON Schema定义）和输出结果。比如，我们会创建一个 search_github_issues 工具，让智能体能去GitHub上搜索问题；再创建一个 query_local_database 工具，让它能查询我们本地的业务数据。

提示（Prompts） 这相当于给智能体预置的“快捷指令”或“对话模板”。你可以提前设计好一些复杂的、多步骤的提示词，用户只需选择一个提示，智能体就会按照预设的流程开始工作。比如，你可以创建一个名为“生成周报”的提示，这个提示内部会引导智能体依次调用“查询本周GitHub提交”、“获取JIRA任务状态”、“总结成邮件草稿”等一系列工具。这对于标准化复杂流程非常有用。

MCP的通信基于 JSON-RPC 2.0 协议，这是一种轻量级的远程过程调用协议。简单来说，客户端（LLM）和服务器（我们的MCP智能体后端）之间通过交换JSON格式的消息来通信。客户端发送一个请求告诉服务器“调用某个工具，参数是这些”，服务器执行后，再把结果以JSON格式返回。这种标准化让不同LLM（Claude Desktop, Cursor, 甚至是自己部署的模型）都能以同样的方式与我们的服务器对话。

3. 从零编写你的第一个MCP服务器

理论说得差不多了，现在让我们打开代码编辑器，开始真正构建东西。我会带你创建一个最简单的MCP服务器，它只做一件事：告诉智能体当前服务器的时间。别小看这个“Hello World”，它能帮你打通整个流程，理解MCP服务器是如何启动、声明能力并处理请求的。

3.1 项目结构与入口文件

在你的项目根目录下，创建如下结构：

my-mcp-agent/
├── venv/               # Python虚拟环境（自动生成）
├── .env                # 环境变量配置文件（稍后创建）
├── .gitignore
├── requirements.txt    # 依赖列表
├── server.py          # 我们的MCP服务器主程序
└── docs/              # 示例资源目录
    └── guide.md

首先，我们来编写 server.py。我将代码拆解成几部分，并加上详细注释。

# server.py
import asyncio
from datetime import datetime
from mcp import ClientSession, StdioServerParameters
from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio

# 创建MCP服务器实例
server = Server("my-first-agent")

# 1. 注册一个简单的工具：获取当前时间
@server.list_tools()
async def handle_list_tools():
    # 返回服务器提供的所有工具列表
    return [
        {
            "name": "get_current_time",
            "description": "获取服务器当前的日期和时间。",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "format": {
                        "type": "string",
                        "description": "时间格式，例如：'%Y-%m-%d %H:%M:%S' 或 'iso'。默认为ISO格式。",
                        "default": "iso"
                    }
                }
            }
        }
    ]

# 2. 处理工具调用请求
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    if name == "get_current_time":
        fmt = arguments.get("format", "iso")
        now = datetime.now()
        
        if fmt == "iso":
            result = now.isoformat()
        else:
            try:
                result = now.strftime(fmt)
            except Exception:
                result = f"错误的时间格式: {fmt}。使用默认ISO格式：{now.isoformat()}"
        
        # 返回工具调用结果，MCP要求返回一个列表
        return [{
            "type": "text",
            "text": f"当前服务器时间是：{result}"
        }]
    else:
        # 如果收到未知工具调用，返回错误
        raise ValueError(f"未知工具: {name}")

# 3. 主异步函数，启动服务器
async def main():
    # 配置服务器使用标准输入输出（stdio）方式通信
    # 这是最简单的方式，适合与Claude Desktop、Cursor等客户端集成
    server_params = StdioServerParameters(
        command="python",  # 解释器
        args=["server.py"] # 脚本参数，这里就是它自己
    )
    
    # 运行服务器，开始监听请求
    async with mcp.server.stdio.stdio_server(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 初始化会话，交换能力信息
            await session.initialize(
                InitializationOptions(
                    server_name="my-first-agent",
                    server_version="0.1.0",
                    capabilities=server.get_capabilities(
                        notification_options=None,
                        experimental_capabilities={},
                    ),
                )
            )
            
            # 将我们定义的工具处理函数注册到会话
            await server.run(
                session,
                InitializationOptions(
                    server_name="my-first-agent",
                    server_version="0.1.0",
                ),
            )

# 程序入口
if __name__ == "__main__":
    asyncio.run(main())

这段代码虽然不长，但包含了一个MCP服务器的所有核心要素。我们创建了一个 Server 对象，然后通过装饰器 @server.list_tools() 声明了这个服务器能提供什么工具（这里只有一个 get_current_time）。当客户端请求调用这个工具时，handle_call_tool 函数会被触发，我们根据参数格式化当前时间并返回。

3.2 运行与测试你的服务器

现在，让我们来测试一下这个服务器是否工作。打开一个终端，确保你在虚拟环境中，然后运行：

python server.py

你会看到程序启动并挂起，等待来自客户端的连接。这说明你的服务器已经在运行了。

接下来，我们需要一个MCP客户端来和它对话。最方便的方式是使用MCP SDK自带的CLI工具。请新开一个终端窗口，激活同一个虚拟环境，然后运行：

# 启动一个交互式会话，连接到我们本地运行的服务器
mcp dev server.py

这个命令会启动一个开发客户端，并连接到 server.py 这个脚本定义的服务器。连接成功后，你应该能看到客户端的提示符。试着输入：

/list_tools

你应该能看到返回的JSON，里面列出了我们定义的 get_current_time 工具。然后，让我们调用它：

/call_tool get_current_time

或者带参数调用：

/call_tool get_current_time '{"format": "%Y年%m月%d日 %H点%M分"}'

如果一切顺利，你会看到服务器返回的当前时间。恭喜你！你已经成功创建了第一个MCP服务器，并完成了与客户端的第一次对话。这个过程虽然简单，但你已经实践了MCP最核心的“工具注册”和“工具调用”流程。有了这个基础，我们就可以往里面添加更强大、更实用的功能了。

4. 赋予智能体“感官”：集成外部数据源与工具

一个只会报时的智能体显然不够看。现在，我们要给它装上“眼睛”和“手”，让它能真正与外界交互。这部分是实战中最有趣也最体现价值的地方。我将以集成 GitHub API 和 本地SQLite数据库 为例，带你一步步实现。

4.1 集成GitHub API，让智能体读懂你的代码仓库

很多开发者的一天是从查看GitHub通知开始的。让我们创建一个工具，让智能体能帮我们查询issue。

首先，我们需要一个GitHub个人访问令牌（Personal Access Token）。去你的GitHub账号设置里生成一个，记得勾上 repo（访问私有仓库）和 read:org 等需要的权限。然后，在项目根目录创建 .env 文件来安全地存储它：

# .env
GITHUB_ACCESS_TOKEN=你的_token_在这里

注意：务必把 .env 添加到 .gitignore 中，千万不要泄露你的Token！

接下来，我们修改 server.py，添加GitHub工具。我们需要安装 httpx 并编写异步函数。

# 在 server.py 顶部添加导入
import os
import httpx
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()
GITHUB_TOKEN = os.getenv("GITHUB_ACCESS_TOKEN")

# 在 handle_list_tools 函数返回的列表中添加新工具
# 修改 @server.list_tools() 对应的函数
@server.list_tools()
async def handle_list_tools():
    return [
        # ... 之前的时间工具 ...
        {
            "name": "search_github_issues",
            "description": "在指定的GitHub仓库中搜索issues。",
            "inputSchema": {
                "type": "object",
                "required": ["owner", "repo"], # 必填参数
                "properties": {
                    "owner": {
                        "type": "string",
                        "description": "仓库所有者的用户名或组织名。"
                    },
                    "repo": {
                        "type": "string",
                        "description": "仓库名称。"
                    },
                    "state": {
                        "type": "string",
                        "description": "issue状态，可选 'open', 'closed', 'all'。",
                        "enum": ["open", "closed", "all"],
                        "default": "open"
                    },
                    "keyword": {
                        "type": "string",
                        "description": "搜索关键词，会在标题和正文中匹配。"
                    }
                }
            }
        }
    ]

# 在 handle_call_tool 函数中添加对新工具的处理
# 修改 @server.call_tool() 对应的函数
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list:
    if name == "get_current_time":
        # ... 原有代码 ...
    elif name == "search_github_issues":
        owner = arguments["owner"]
        repo = arguments["repo"]
        state = arguments.get("state", "open")
        keyword = arguments.get("keyword", "")
        
        # 构建GitHub API查询URL
        query = f"repo:{owner}/{repo} is:issue state:{state}"
        if keyword:
            query += f" {keyword} in:title,body"
        
        url = "https://api.github.com/search/issues"
        headers = {
            "Authorization": f"token {GITHUB_TOKEN}",
            "Accept": "application/vnd.github.v3+json"
        }
        params = {"q": query, "per_page": 5} # 限制返回5条
        
        async with httpx.AsyncClient() as client:
            try:
                response = await client.get(url, headers=headers, params=params)
                response.raise_for_status() # 如果状态码不是2xx，抛出异常
                data = response.json()
                
                items = data.get("items", [])
                if not items:
                    return [{"type": "text", "text": f"在 {owner}/{repo} 中没有找到匹配的issue。"}]
                
                # 格式化输出
                summary = []
                for item in items:
                    summary.append(f"* [#{item['number']}] {item['title']} - {item['html_url']}")
                
                result_text = f"在 {owner}/{repo} 中找到 {len(items)} 个issue：\n" + "\n".join(summary)
                return [{"type": "text", "text": result_text}]
                
            except httpx.HTTPStatusError as e:
                return [{"type": "text", "text": f"GitHub API请求失败，状态码：{e.response.status_code}"}]
            except Exception as e:
                return [{"type": "text", "text": f"请求发生错误：{str(e)}"}]
    else:
        raise ValueError(f"未知工具: {name}")

现在，重启你的服务器 (python server.py)，并在测试客户端里调用新工具：

/call_tool search_github_issues '{"owner": "openai", "repo": "openai-python", "state": "open"}'

你应该能看到OpenAI Python SDK仓库里最近的一些open issue被列了出来。看，你的智能体现在已经能“看到”GitHub的世界了！

4.2 连接本地数据库，让智能体掌握业务数据

对于企业应用，连接内部数据库是刚需。我们以轻量级的SQLite为例，演示如何让智能体查询业务数据。假设我们有一个简单的产品表。

首先，创建一个 database.py 文件来初始化数据库和表结构：

# database.py
import sqlite3

def init_database():
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()
    
    # 创建一个产品表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            category TEXT,
            price REAL,
            stock INTEGER
        )
    ''')
    
    # 插入一些示例数据
    cursor.execute("DELETE FROM products") # 清空旧数据，仅演示用
    sample_products = [
        ('无线蓝牙耳机', '电子产品', 299.99, 150),
        ('编程思想入门', '图书', 59.80, 30),
        ('不锈钢保温杯', '生活用品', 89.00, 200),
        ('Python机器学习', '图书', 108.50, 25),
        ('智能手环', '电子产品', 399.00, 80),
    ]
    cursor.executemany('INSERT INTO products (name, category, price, stock) VALUES (?, ?, ?, ?)', sample_products)
    
    conn.commit()
    conn.close()
    print("数据库初始化完成。")

if __name__ == "__main__":
    init_database()

运行一次 python database.py 来创建数据库和示例数据。

然后，我们在 server.py 中增加数据库查询工具。这里使用 sqlite3 标准库，注意MCP服务器是异步的，而 sqlite3 是同步的，为了避免阻塞事件循环，我们使用 asyncio.to_thread 在单独线程中执行数据库操作。

# 在 server.py 顶部添加导入
import sqlite3
import asyncio

# 在 handle_list_tools 返回的列表中添加新工具
        {
            "name": "query_products",
            "description": "查询产品数据库，支持按名称、类别筛选，或列出库存紧张的产品。",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "category": {
                        "type": "string",
                        "description": "按产品类别筛选，例如：'电子产品'、'图书'。"
                    },
                    "name_keyword": {
                        "type": "string",
                        "description": "在产品名称中搜索关键词。"
                    },
                    "low_stock_threshold": {
                        "type": "integer",
                        "description": "列出库存低于此阈值的产品。如果不提供，则不过滤库存。"
                    }
                }
            }
        }

# 在 handle_call_tool 中添加处理逻辑
    elif name == "query_products":
        category = arguments.get("category")
        name_keyword = arguments.get("name_keyword")
        low_stock = arguments.get("low_stock_threshold")
        
        # 构建SQL查询
        query = "SELECT name, category, price, stock FROM products WHERE 1=1"
        params = []
        
        if category:
            query += " AND category = ?"
            params.append(category)
        if name_keyword:
            query += " AND name LIKE ?"
            params.append(f'%{name_keyword}%')
        if low_stock is not None:
            query += " AND stock < ?"
            params.append(low_stock)
        
        query += " ORDER BY stock ASC" # 按库存升序排列
        
        # 在独立线程中执行同步的数据库操作
        def sync_db_query():
            conn = sqlite3.connect('example.db')
            conn.row_factory = sqlite3.Row # 以字典形式返回行
            cursor = conn.cursor()
            cursor.execute(query, params)
            rows = cursor.fetchall()
            conn.close()
            return rows
        
        try:
            rows = await asyncio.to_thread(sync_db_query)
            
            if not rows:
                return [{"type": "text", "text": "没有找到匹配的产品。"}]
            
            # 格式化结果
            product_list = []
            for row in rows:
                product_list.append(f"- {row['name']} ({row['category']})， 价格：¥{row['price']}， 库存：{row['stock']}件")
            
            result_text = f"找到 {len(rows)} 个产品：\n" + "\n".join(product_list)
            return [{"type": "text", "text": result_text}]
            
        except sqlite3.Error as e:
            return [{"type": "text", "text": f"数据库查询错误：{str(e)}"}]

重启服务器后，试试这些命令：

/call_tool query_products '{"category": "图书"}'
/call_tool query_products '{"low_stock_threshold": 50}'

智能体现在可以查询你的“业务数据”了。通过这两个例子，你已经掌握了集成API和数据库的精髓。同样的模式，你可以扩展到任何外部系统：发送邮件的SMTP、查询天气的API、控制智能家居的IoT平台等等。你的智能体正在变得无所不能。

5. 连接大脑：将MCP服务器与LLM客户端对接

服务器准备好了，工具也齐全了，现在我们需要为它找一个“大脑”——一个能够理解自然语言、并懂得调用我们这些工具的LLM。这里有几个主流的选择，我会分别介绍如何对接，你可以根据自己的情况选择。

5.1 对接Claude Desktop（最便捷的方式）

如果你使用的是Anthropic的Claude Desktop应用，那么集成起来是最简单的。Claude原生支持MCP。你只需要创建一个配置文件，告诉Claude你的服务器在哪里。

在特定目录下创建MCP服务器配置文件。这个目录位置因操作系统而异：

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
• Windows: %APPDATA%\Claude\claude_desktop_config.json
• Linux: ~/.config/Claude/claude_desktop_config.json

编辑这个JSON文件（如果不存在就创建）：

{
  "mcpServers": {
    "my-local-agent": {
      "command": "/绝对/路径/到/你的/venv/bin/python",
      "args": ["/绝对/路径/到/你的/my-mcp-agent/server.py"],
      "env": {
        "GITHUB_ACCESS_TOKEN": "你的_token"
      }
    }
  }
}

注意：command 必须是你虚拟环境中Python解释器的绝对路径。args 是你的 server.py 的绝对路径。env 部分可以传递环境变量，这样就不用在代码里加载 .env 文件了（更安全）。

保存配置，重启Claude Desktop。重启后，在聊天界面，你应该能看到Claude的回复中偶尔会出现一个微小的齿轮图标，或者当你输入相关指令时（如“帮我查一下OpenAI仓库的issue”），Claude会自动识别并调用你服务器里的工具。你可以要求Claude“列出可用的工具”来确认连接成功。

5.2 对接Cursor IDE（开发者的利器）

Cursor是另一款深度集成AI的代码编辑器，它也支持MCP。在Cursor中，你可以通过命令面板快速连接。

1. 在Cursor中打开你的项目文件夹。
2. 按下 Cmd/Ctrl + Shift + P 打开命令面板。
3. 输入 MCP: Add New Server 并选择。
4. 选择 Command Line 类型。
5. 在 Command 栏填入你的Python解释器路径（虚拟环境内的）。
6. 在 Args 栏填入 server.py 的路径。
7. 保存后，Cursor就会启动你的MCP服务器。在聊天框中，你可以直接使用自然语言，比如“用我的工具查一下产品库存”，Cursor背后的AI（通常是GPT）就会去调用对应的工具。

5.3 对接自定义LLM应用（最灵活的方式）

如果你想在自己的应用里使用这个MCP智能体，比如集成一个开源的Llama 3模型，你需要使用MCP的客户端库。这里给出一个使用官方 mcp Python库编写简单客户端的例子：

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

async def main():
    # 配置连接到我们自己的服务器
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"]
    )
    
    async with mcp.client.stdio.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])
            
            # 假设我们有一个LLM，它分析用户输入后决定调用 `query_products`
            # 这里我们模拟LLM的输出，直接调用
            result = await session.call_tool(
                "query_products",
                arguments={"category": "电子产品"}
            )
            print("查询结果:", result.content[0].text)
            
            # 你可以在这里将结果发送给真正的LLM，让它生成友好的回复给用户

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

这个自定义客户端充当了LLM和MCP服务器之间的桥梁。你的LLM应用逻辑是：用户输入 -> LLM分析意图并决定调用哪个工具及参数 -> 通过这个客户端调用MCP服务器 -> 获取结果 -> LLM将结果组织成自然语言回复给用户。

5.4 调试与问题排查

在对接过程中，你可能会遇到连接失败、工具不显示等问题。这里分享几个我踩过的坑和排查技巧：

1. 路径问题：Claude/Cursor的配置文件里，command 和 args 的路径必须是绝对路径，并且确保该Python环境已安装所有依赖（mcp, httpx等）。
2. 权限问题：确保你的脚本有可执行权限，并且环境变量配置正确。可以在终端手动运行配置中的命令来测试服务器是否能独立启动。
3. 查看日志：Claude Desktop在连接MCP服务器时，会在上述配置文件的同级目录生成日志文件，查看 claude_desktop_log.txt 可以找到具体的错误信息。
4. 使用MCP CLI调试：就像我们之前做的，mcp dev server.py 是一个极佳的调试工具，可以手动测试工具调用是否正常。
5. 服务器保持运行：确保你的MCP服务器脚本是持续运行的守护进程，而不是执行一次就退出。我们的示例代码使用了 async with 和事件循环，会一直运行。

当你看到LLM能够流畅地使用你定义的工具时，那种成就感是无与伦比的。你不再是单纯地在和模型聊天，而是在指挥一个拥有真实“手脚”的智能助手。

6. 进阶实战：构建多步骤任务与上下文管理

基础功能跑通后，我们的智能体已经能处理单次请求了。但现实世界的任务往往是复杂的、多步骤的。比如，“查一下仓库里的bug，如果超过10个就发邮件提醒我”。这需要智能体具备规划和记忆能力。MCP协议本身不直接提供复杂的Agent工作流引擎，但它提供了完美的底层支持。我们可以结合一些设计模式来实现。

6.1 设计复合工具与工作流

与其期待LLM自己完美地规划多步步骤，不如我们将常见的复杂流程封装成更高级的“复合工具”。例如，我们可以创建一个“生成项目状态报告”的工具。

这个工具内部会按顺序做以下几件事：

1. 调用 search_github_issues 获取最新的open issue。
2. 调用 query_products 获取库存紧张的产品。
3. 将两部分信息汇总，格式化为一份报告。

# 在 server.py 的 handle_list_tools 中添加新工具
        {
            "name": "generate_project_status",
            "description": "生成一份简要的项目状态报告，包括GitHub issue情况和产品库存状态。",
            "inputSchema": {
                "type": "object",
                "required": ["github_owner", "github_repo"],
                "properties": {
                    "github_owner": {"type": "string"},
                    "github_repo": {"type": "string"},
                    "low_stock_threshold": {
                        "type": "integer",
                        "default": 50
                    }
                }
            }
        }

# 在 handle_call_tool 中实现它
    elif name == "generate_project_status":
        github_owner = arguments["github_owner"]
        github_repo = arguments["github_repo"]
        low_stock = arguments.get("low_stock_threshold", 50)
        
        report_parts = []
        
        # 1. 获取GitHub issue (复用之前的逻辑，这里简化为直接调用内部函数)
        # 注意：实际项目中，更好的设计是将工具函数独立出来，供复合工具调用
        issues_text = await handle_github_issue_search(github_owner, github_repo) # 假设这个函数存在
        report_parts.append(f"## GitHub 仓库状态 ({github_owner}/{github_repo})")
        report_parts.append(issues_text)
        
        # 2. 获取低库存产品
        products_text = await handle_low_stock_query(low_stock) # 假设这个函数存在
        report_parts.append(f"\n## 产品库存警报 (库存 < {low_stock})")
        report_parts.append(products_text)
        
        # 3. 生成最终报告
        final_report = "\n".join(report_parts)
        return [{"type": "text", "text": final_report}]

这样，用户或LLM只需要调用这一个 generate_project_status 工具，就能得到一个组合了多个信息源的综合报告。这是简化复杂任务的第一步。

6.2 利用会话与LLM的上下文进行管理

对于无法预定义的、动态的多步骤任务，我们需要依赖LLM自身的推理和规划能力。MCP协议支持会话（Session），在一个会话中，LLM可以记住之前的工具调用和结果，并据此决定下一步行动。

这就是 ReAct（Reasoning + Acting） 模式的用武之地。你不需要在MCP服务器端实现它，而是在你的客户端或LLM的提示词（Prompt）中引导模型进行这种“思考-行动”的循环。

例如，你可以给LLM这样的系统提示词：

你是一个强大的助手，可以调用各种工具来帮助用户。在回答问题时，请遵循以下步骤：
1. 思考：分析用户请求，判断是否需要调用工具，以及需要哪些信息。
2. 行动：如果需要，调用合适的工具。一次只调用一个工具。
3. 观察：获取工具返回的结果。
4. 循环：根据结果，决定是继续调用其他工具，还是已经有了足够信息来生成最终答案。
你可以调用的工具包括：[列出所有工具名称和描述]。

当用户问“我仓库里最紧急的bug是什么，对应的产品库存还够吗？”，LLM可能会这样工作：

1. 思考：需要先查issue，找到最紧急的（比如最近创建的或标记为bug的）。
2. 行动：调用 search_github_issues，参数为 {“state”: “open”, “keyword”: “bug”}。
3. 观察：得到issue列表。
4. 思考：从结果中识别出关键产品名，比如“无线蓝牙耳机连接问题”。
5. 行动：调用 query_products，参数为 {“name_keyword”: “无线蓝牙耳机”}。
6. 观察：得到该产品的库存。
7. 最终回答：将两部分信息综合，告诉用户“最紧急的bug是XXX，涉及的产品当前库存是YYY，建议...”

MCP服务器在这个过程中，只是忠实地提供了一个个独立的工具。而任务规划、上下文串联的工作，交给了更擅长此道的LLM。这种分工非常清晰高效。

6.3 错误处理与健壮性提升

在实际使用中，网络会波动，API会限流，数据库会超时。一个健壮的智能体必须能妥善处理这些情况。

1. 工具层面的错误处理：我们已经在GitHub和数据库工具中使用了 try...except 块来捕获异常，并返回友好的错误信息给LLM，而不是让整个会话崩溃。这很重要，因为LLM需要知道“工具调用失败了”，而不是得到一个空响应或崩溃。
2. 参数验证：MCP利用JSON Schema在调用前就进行初步的参数验证（类型、必填项、枚举值等）。但我们仍应在工具函数内部进行业务逻辑验证，比如检查GitHub仓库是否存在。
3. 设置超时：对于网络请求，务必设置超时。在 httpx.AsyncClient 中，可以配置 timeout=30.0。
4. 重试机制：对于可能因瞬时网络问题失败的API调用，可以实现简单的重试逻辑。但要小心，不是所有失败都适合重试（比如认证失败）。
5. 结果格式化：返回给LLM的结果应该清晰、结构化。避免返回巨大的JSON原始数据。像我们之前做的那样，将数据总结成易于理解的文本段落，能极大提升LLM生成优质回答的几率。

7. 部署与展望：让你的智能体服务更多人

开发调试完成后，你可能希望将这个智能体部署到服务器上，供团队或更多人稳定使用。部署一个MCP服务器和部署一个普通的Web服务没有本质区别，但有一些细节需要注意。

7.1 部署选项与配置

方案一：作为长期运行的服务进程 这是最常见的方式。你可以使用 systemd (Linux), launchd (macOS) 或 NSSM (Windows) 将你的 server.py 脚本注册为系统服务，让它开机自启，并在崩溃时自动重启。你需要确保服务运行在正确的Python虚拟环境下，并且所有环境变量都已配置。

一个简单的 systemd 服务文件示例 (/etc/systemd/system/my-mcp-agent.service)：

[Unit]
Description=My MCP Agent Server
After=network.target

[Service]
Type=simple
User=your_username
WorkingDirectory=/path/to/your/my-mcp-agent
Environment="PATH=/path/to/your/my-mcp-agent/venv/bin"
Environment="GITHUB_ACCESS_TOKEN=your_token_here"
ExecStart=/path/to/your/my-mcp-agent/venv/bin/python /path/to/your/my-mcp-agent/server.py
Restart=on-failure
RestartSec=5s

[Install]
WantedBy=multi-user.target

方案二：容器化部署（推荐） 使用Docker可以将你的应用及其所有依赖打包成一个镜像，确保在任何环境下的运行一致性。创建一个 Dockerfile：

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
# 初始化数据库（根据实际情况调整）
RUN python database.py
CMD ["python", "server.py"]

然后构建镜像并运行容器。在云服务器上，结合Docker Compose或Kubernetes可以轻松管理。

方案三：集成到现有Web框架 如果你的应用本身是一个Web服务（比如FastAPI），你可以在其中启动MCP服务器，并通过WebSocket或HTTP端点对外提供MCP服务。这需要更复杂的集成，但灵活性最高。mcp 库提供了底层的通信组件，可以嵌入到任何异步框架中。

7.2 安全与权限考量

一旦部署，安全就是头等大事。

• 令牌管理：绝对不要将API令牌、数据库密码等硬编码在代码或镜像中。使用环境变量、密钥管理服务（如AWS Secrets Manager, HashiCorp Vault）或容器编排平台的Secret功能。
• 网络隔离：确保你的MCP服务器运行在内部网络，不要直接暴露在公网。客户端（如Claude Desktop）通过本地回环地址或安全的内部网络连接到它。
• 工具权限：在工具定义时，要深思熟虑。提供文件写入工具？那就要严格限制可访问的目录。提供数据库删除工具？那可能就需要额外的确认机制，或者干脆不提供。MCP协议支持服务器向客户端发送权限请求，你可以利用这一点在执行危险操作前进行二次确认。
• 输入验证与清理：对所有来自外部的输入（包括工具参数）进行严格的验证和清理，防止注入攻击，特别是在拼接SQL或系统命令时。

7.3 性能监控与日志

为了让你的智能体稳定运行，你需要观察它。

• 日志记录：在服务器代码中添加详细的日志记录，记录工具调用、参数、结果和错误。可以使用Python标准库的 logging 模块，配置输出到文件和控制台。
• 性能指标：考虑记录每个工具调用的耗时，这有助于你发现性能瓶颈。
• 健康检查：可以添加一个简单的健康检查工具（如 ping），让监控系统定期调用，以确保服务器存活。

走到这一步，你已经拥有了一个功能完整、可部署的MCP智能体。它不再是一个玩具，而是一个可以融入你或团队工作流的生产力工具。从最初的一个简单报时工具，到如今能连接内外数据、处理复杂请求的智能助手，这个过程本身就是一个极佳的学习和创造之旅。MCP的魅力在于它的简洁和标准化，它没有试图解决所有问题，而是提供了一个优雅的接口，让强大的LLM和丰富的现实世界得以安全、高效地连接。未来，你可以继续为它添加更多工具，比如连接日历、管理待办事项、控制智能家居，真正打造一个属于你的数字生活副驾驶。