API升级MCP教程2026
快速将现有API升级为MCP供AI-Agent使用 + 好用MCP开源项目清单

# Python（推荐）
pip install "mcp[cli]" httpx

# Node.js
npm install @modelcontextprotocol/sdk

# 原有图书API（FastAPI）
GET  /books              → 获取图书列表
GET  /books/{id}         → 获取单本图书详情
POST /books              → 新增图书
DELETE /books/{id}       → 删除图书

"""
mcp_book_server.py
将现有图书API快速升级为MCP Server
"""

import json
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# ========== 配置区 ==========
BOOK_API_BASE = "http://localhost:8000"   # 现有图书API地址
API_KEY = "your-api-key-here"              # 现有API认证密钥

# ========== MCP Server初始化 ==========
server = Server("book-mcp-server")

# ========== 工具定义（AI自动发现） ==========
TOOLS = [
    Tool(
        name="list_books",
        description="获取图书列表，支持分页和关键词搜索",
        inputSchema={
            "type": "object",
            "properties": {
                "page": {"type": "integer", "description": "页码", "default": 1},
                "page_size": {"type": "integer", "description": "每页数量", "default": 20},
                "keyword": {"type": "string", "description": "搜索关键词（书名/作者）"},
            },
        },
    ),
    Tool(
        name="get_book",
        description="根据ID获取单本图书详情",
        inputSchema={
            "type": "object",
            "properties": {
                "book_id": {"type": "string", "description": "图书唯一标识"},
            },
            "required": ["book_id"],
        },
    ),
    Tool(
        name="add_book",
        description="新增一本图书",
        inputSchema={
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "书名"},
                "author": {"type": "string", "description": "作者"},
                "isbn": {"type": "string", "description": "ISBN"},
                "price": {"type": "number", "description": "价格（元）"},
            },
            "required": ["title", "author"],
        },
    ),
    Tool(
        name="delete_book",
        description="根据ID删除图书",
        inputSchema={
            "type": "object",
            "properties": {
                "book_id": {"type": "string", "description": "图书唯一标识"},
            },
            "required": ["book_id"],
        },
    ),
]

# ========== 工具执行器（HTTP → MCP桥接） ==========
async def call_tool(name: str, arguments: dict):
    headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
    async with httpx.AsyncClient(timeout=30.0) as client:
        if name == "list_books":
            params = {"page": arguments.get("page", 1), "page_size": arguments.get("page_size", 20)}
            if arguments.get("keyword"):
                params["keyword"] = arguments["keyword"]
            resp = await client.get(f"{BOOK_API_BASE}/books", headers=headers, params=params)
            return TextContent(type="text", text=json.dumps(resp.json(), ensure_ascii=False, indent=2))
        elif name == "get_book":
            resp = await client.get(f"{BOOK_API_BASE}/books/{arguments['book_id']}", headers=headers)
            return TextContent(type="text", text=json.dumps(resp.json(), ensure_ascii=False, indent=2))
        elif name == "add_book":
            resp = await client.post(f"{BOOK_API_BASE}/books", headers=headers, json={
                "title": arguments["title"], "author": arguments["author"],
                "isbn": arguments.get("isbn"), "price": arguments.get("price"),
            })
            return TextContent(type="text", text=json.dumps(resp.json(), ensure_ascii=False, indent=2))
        elif name == "delete_book":
            resp = await client.delete(f"{BOOK_API_BASE}/books/{arguments['book_id']}", headers=headers)
            return TextContent(type="text", text=f"删除成功，状态码: {resp.status_code}")
        else:
            return TextContent(type="text", text=f"未知工具: {name}")

# ========== MCP协议处理 ==========
@server.list_tools()
async def list_tools():
    return TOOLS

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict):
    result = await call_tool(name, arguments)
    return [result]

# ========== 启动MCP Server（stdio模式） ==========
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

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

{
  "mcpServers": {
    "book-mcp-server": {
      "command": "python3",
      "args": ["/path/to/mcp_book_server.py"],
      "env": {}
    }
  }
}

# 将MCP Server部署为HTTP+SSE模式，多个Agent可并发接入
from mcp.server.sse import sse_server

async def main_sse():
    """HTTP+SSE模式，监听8080端口"""
    async with sse_server(app, port=8080) as streams:
        await server.run(streams[0], streams[1], server.create_initialization_options())

# Claude Desktop配置改为：
# {
#   "mcpServers": {
#     "book-mcp-server": {
#       "url": "http://localhost:8080/sse"
#     }
#   }
# }