/ MCP  AI Agent  Claude  大模型  API集成  Python  LLM开发  工具开发 

MCP 工具开发实战:从零给 AI 接入任意 API 的完整指南


文章封面

MCP(Model Context Protocol)是 Anthropic 在 2024 年底发布的开放协议,旨在让 AI 助手能够安全、标准化地接入任意外部 API 和数据源。简单理解:它就像给 AI 装上了"插件系统"——你写一个 MCP Server,AI 就能调用你提供的工具,无论是查数据库、调用内部 API,还是操控本地文件系统,都能一把搞定。

本文以一个真实场景为例:为 AI 助手开发一个能查询公司内部项目状态的 MCP Server,从零开始,覆盖协议原理、开发环境搭建、工具定义、鉴权处理、调试技巧到实际接入 Claude Desktop,让你看完就能动手。

一、MCP 是什么?为什么你需要它

在 MCP 出现之前,想让 AI 接入外部系统,每个平台都要单独实现"Function Calling"或"Tool Use"——OpenAI 有一套格式,Claude 有一套格式,各家 Agent 框架又各有不同。开发者苦不堪言,要么为每个平台单独维护集成代码,要么依赖某个框架的封装(但框架升级就可能崩)。

MCP 的目标是统一这个混乱局面。它定义了一套标准的 Client-Server 通信协议:

  • MCP Server:你开发的工具提供方,暴露一组可调用的"工具"(Tools)、可读取的"资源"(Resources)和可使用的"提示词"(Prompts)

  • MCP Client:集成在 AI 助手中的协议客户端,负责发现、调用 MCP Server 提供的能力

  • 传输层:支持 stdio(本地进程通信)和 HTTP+SSE(远程服务)两种方式

目前支持 MCP 的平台已经相当丰富:Claude Desktop、Cursor、Continue、Cline、以及各种基于 LangChain/LlamaIndex 构建的 Agent 框架。一次开发,处处可用。

回到我们的场景:公司有一套内部项目管理系统,提供 REST API,但 Claude Desktop 没法直接查它。通过 MCP,我们可以让 AI 直接回答"目前有哪些进行中的项目"、"张三负责哪些任务"这类问题,而不需要用户手动去系统里查再粘贴给 AI。

二、开发环境搭建

MCP SDK 目前官方支持 Python 和 TypeScript,本文使用 Python(更适合快速原型)。

# 推荐用 uv 管理 Python 环境(比 pip + venv 更快更干净)
curl -LsSf https://astral.sh/uv/install.sh | sh

# 创建项目
mkdir mcp-project-server && cd mcp-project-server
uv init --python 3.11
uv add mcp httpx python-dotenv

# 目录结构
.
├── .env                 # 存放 API Token(不提交到 git)
├── pyproject.toml
└── src/
    └── server.py        # MCP Server 主文件

关于 Python 版本:MCP SDK 要求 Python 3.10+,推荐 3.11 或 3.12,3.13 目前个别依赖还有兼容问题。

创建 .env 文件,存放内部系统的 API 配置:

PROJECT_API_BASE=https://internal.company.com/api/v2
PROJECT_API_TOKEN=your_secret_token_here

三、核心代码:定义你的第一个 MCP 工具

MCP Server 的核心是定义"工具"——每个工具对应一个 AI 可以调用的操作,类似 OpenAI 的 Function Calling,但格式更标准化。

# src/server.py
import asyncio
import os
from dotenv import load_dotenv
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

load_dotenv()

API_BASE = os.getenv("PROJECT_API_BASE")
API_TOKEN = os.getenv("PROJECT_API_TOKEN")

app = Server("project-server")

# 定义工具列表(AI 会根据这些描述决定什么时候调用哪个工具)
@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="list_projects",
            description="列出公司内部项目列表,可按状态筛选(进行中/已完成/暂停)",
            inputSchema={
                "type": "object",
                "properties": {
                    "status": {
                        "type": "string",
                        "enum": ["active", "completed", "paused", "all"],
                        "description": "项目状态筛选,默认为 all",
                        "default": "all"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "返回数量上限,默认 20",
                        "default": 20
                    }
                }
            }
        ),
        types.Tool(
            name="get_project_detail",
            description="获取指定项目的详细信息,包括任务列表、负责人、进度等",
            inputSchema={
                "type": "object",
                "properties": {
                    "project_id": {
                        "type": "string",
                        "description": "项目 ID"
                    }
                },
                "required": ["project_id"]
            }
        ),
        types.Tool(
            name="search_tasks_by_assignee",
            description="查询某个员工负责的所有任务",
            inputSchema={
                "type": "object",
                "properties": {
                    "assignee": {
                        "type": "string",
                        "description": "员工姓名或工号"
                    },
                    "status": {
                        "type": "string",
                        "enum": ["todo", "in_progress", "done", "all"],
                        "default": "all"
                    }
                },
                "required": ["assignee"]
            }
        )
    ]

工具描述(description)非常关键——AI 就是根据这段文字来判断什么时候应该调用这个工具的。描述要清晰、具体,说清楚这个工具能做什么、适合什么场景。模糊的描述会导致 AI 乱调或不调。

四、实现工具逻辑:对接真实 API

工具定义好之后,需要实现实际的调用逻辑。当 AI 决定使用某个工具时,MCP 框架会触发 call_tool 处理器:

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    headers = {
        "Authorization": f"Bearer {API_TOKEN}",
        "Content-Type": "application/json"
    }
    
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            if name == "list_projects":
                status = arguments.get("status", "all")
                limit = arguments.get("limit", 20)
                
                params = {"limit": limit}
                if status != "all":
                    params["status"] = status
                
                resp = await client.get(
                    f"{API_BASE}/projects",
                    headers=headers,
                    params=params
                )
                resp.raise_for_status()
                data = resp.json()
                
                # 格式化输出——让 AI 更容易理解
                projects = data.get("items", [])
                if not projects:
                    return [types.TextContent(type="text", text="没有找到符合条件的项目")]
                
                lines = [f"共找到 {len(projects)} 个项目:\n"]
                for p in projects:
                    lines.append(
                        f"- [{p['id']}] {p['name']} | 状态: {p['status']} | "
                        f"负责人: {p.get('owner', '未分配')} | "
                        f"截止日期: {p.get('due_date', '未设置')}"
                    )
                return [types.TextContent(type="text", text="\n".join(lines))]
            
            elif name == "get_project_detail":
                project_id = arguments["project_id"]
                resp = await client.get(
                    f"{API_BASE}/projects/{project_id}",
                    headers=headers
                )
                resp.raise_for_status()
                p = resp.json()
                
                tasks = p.get("tasks", [])
                task_summary = "\n".join([
                    f"  - {t['title']} [{t['status']}] 负责人: {t.get('assignee', '未分配')}"
                    for t in tasks[:10]  # 最多显示10条
                ])
                
                result = f"""项目详情:{p['name']} (ID: {p['id']})
状态: {p['status']}
描述: {p.get('description', '无')}
开始日期: {p.get('start_date', '未设置')}
截止日期: {p.get('due_date', '未设置')}
项目负责人: {p.get('owner', '未分配')}
完成进度: {p.get('progress', 0)}%

任务列表 (共{len(tasks)}条,显示前10条):
{task_summary}"""
                
                return [types.TextContent(type="text", text=result)]
            
            elif name == "search_tasks_by_assignee":
                assignee = arguments["assignee"]
                status = arguments.get("status", "all")
                
                params = {"assignee": assignee}
                if status != "all":
                    params["status"] = status
                
                resp = await client.get(
                    f"{API_BASE}/tasks/search",
                    headers=headers,
                    params=params
                )
                resp.raise_for_status()
                data = resp.json()
                
                tasks = data.get("items", [])
                if not tasks:
                    return [types.TextContent(type="text", text=f"未找到 {assignee} 的任务")]
                
                lines = [f"{assignee} 共有 {len(tasks)} 个任务:\n"]
                for t in tasks:
                    lines.append(
                        f"- [{t['project_name']}] {t['title']} | "
                        f"状态: {t['status']} | 截止: {t.get('due_date', '未设置')}"
                    )
                return [types.TextContent(type="text", text="\n".join(lines))]
            
            else:
                return [types.TextContent(type="text", text=f"未知工具: {name}")]
        
        except httpx.HTTPStatusError as e:
            return [types.TextContent(
                type="text",
                text=f"API 调用失败: HTTP {e.response.status_code} - {e.response.text[:200]}"
            )]
        except httpx.TimeoutException:
            return [types.TextContent(type="text", text="请求超时,内部系统可能暂时不可用")]
        except Exception as e:
            return [types.TextContent(type="text", text=f"意外错误: {str(e)}")]


# 启动 Server
if __name__ == "__main__":
    asyncio.run(stdio_server(app))

有几个踩坑点值得注意:

  • 超时必须设置:内部 API 有时会卡住,不设超时会导致 MCP 连接假死

  • 错误信息要对 AI 友好:不要直接抛异常,返回清晰的文字说明,让 AI 能告知用户具体出了什么问题

  • 输出格式要清晰:AI 会把工具返回的文本直接作为上下文,格式越清晰,AI 理解越准确,回答越好

五、本地调试:不依赖任何 AI 客户端

开发过程中频繁切换到 Claude Desktop 测试非常低效。MCP SDK 提供了命令行调试工具,可以直接在终端验证工具是否正常工作:

# 方法一:使用 mcp dev 工具(推荐)
uv run mcp dev src/server.py

# 这会启动一个交互式 Inspector,在浏览器中打开 http://localhost:5173
# 可以直接列出工具、手动填参数调用、查看返回结果

如果没有图形界面,也可以用脚本直接测试:

# 方法二:写测试脚本
# test_server.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def test():
    server_params = StdioServerParameters(
        command="uv",
        args=["run", "src/server.py"],
        env=None
    )
    
    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("可用工具:")
            for t in tools.tools:
                print(f"  - {t.name}: {t.description}")
            
            # 调用工具测试
            result = await session.call_tool(
                "list_projects",
                arguments={"status": "active", "limit": 5}
            )
            print("\n调用结果:")
            for content in result.content:
                print(content.text)

asyncio.run(test())

调试阶段的常见问题:

  • Server 启动失败:99% 是 .env 没有正确加载,加一行 print(os.getenv("PROJECT_API_BASE")) 确认

  • 工具列表为空:检查 @app.list_tools() 装饰器是否正确,函数签名是否匹配

  • 工具调用报错但无堆栈:MCP 会吞掉未捕获的异常,记得在 call_tool 里加宽泛的 try/except 并打印

六、接入 Claude Desktop:最后一步

代码调通之后,接入 Claude Desktop 只需修改一个配置文件。找到 Claude Desktop 的配置文件:

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

  • Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "project-server": {
      "command": "uv",
      "args": [
        "run",
        "--project",
        "/path/to/mcp-project-server",
        "python",
        "src/server.py"
      ],
      "env": {
        "PROJECT_API_BASE": "https://internal.company.com/api/v2",
        "PROJECT_API_TOKEN": "your_secret_token_here"
      }
    }
  }
}

保存后完全退出并重启 Claude Desktop(注意是完全退出,不是最小化)。重启后在对话框左下角会出现工具图标,点击可以看到你注册的工具列表。

现在你可以直接问 Claude:"帮我看看目前有哪些进行中的项目,以及张三负责什么任务?"它会自动决定调用哪些工具,汇总结果给你。

几个配置的注意事项:

  • 用 uv run 而不是 python:确保使用项目虚拟环境中的依赖,避免包版本冲突

  • env 中的变量会覆盖系统环境变量:不同项目可以有不同的 Token,互不干扰

  • 路径用绝对路径:相对路径在 Claude Desktop 启动时可能解析错误

七、进阶:添加 Resources 和 Prompts

除了 Tools,MCP 还支持 Resources(让 AI 读取数据,如文档、日志)和 Prompts(预定义的提示词模板)。在我们的项目中,可以加一个 Resource,让 AI 直接读取项目的 README:

from mcp import types
from mcp.server import Server

@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="project://docs/workflow",
            name="项目工作流规范",
            description="公司内部项目管理流程和规范文档",
            mimeType="text/markdown"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "project://docs/workflow":
        # 从内部 API 或文件读取
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                f"{API_BASE}/docs/workflow",
                headers={"Authorization": f"Bearer {API_TOKEN}"}
            )
            return resp.text
    raise ValueError(f"未知资源: {uri}")

Resources 和 Tools 的区别:Resources 是 AI "主动读取"的数据,Tools 是 AI "主动调用"的操作。前者适合静态或半静态的内容(文档、配置),后者适合需要参数的动态查询。

八、踩坑总结与最佳实践

经过实际项目开发,总结了以下关键经验:

  • 工具数量不要太多:超过 10-15 个工具,AI 的工具选择准确率会下降。相关功能尽量合并,用参数区分

  • 描述比代码更重要:Tool 的 description 写得好,AI 调用准确率能提升 30%+。花时间打磨描述

  • 返回结构化文本而非 JSON:虽然 JSON 看起来更"专业",但 AI 处理自然语言描述更高效,报错也更清晰

  • 做好降级处理:API 不可用时,返回友好提示而非让 MCP 崩溃,用户体验差很多

  • 注意敏感信息:Token、密码不要打印到 stdout(MCP stdio 模式下 stdout 就是通信通道),用 stderr 打印日志

  • 版本锁定mcp SDK 还在快速迭代,pyproject.toml 中锁定精确版本避免升级破坏

MCP 生态目前还在高速成长,官方 Server 仓库(github.com/modelcontextprotocol/servers)已经有几十个现成实现可以参考,包括 GitHub、Slack、数据库等常见场景。如果你的需求和现有实现相似,直接 fork 改改比从零写更快。

最后:MCP 的价值不仅是技术上的标准化,更是工作方式的改变——以后不用打开十几个系统查信息,直接问 AI 就好了。

发布评论

热门评论区: