Python MCP Server開發：2026年從零建構AI Agent可呼叫的工具服務

AI与大数据

Python MCP Server開發：2026年從零建構AI Agent可呼叫的工具服務

你在開發AI Agent時是不是發現：每個AI平台都有自己的工具呼叫協定，Claude用一套，ChatGPT用另一套，Gemini又不一樣？你的工具程式碼要為每個平台寫一遍適配層？2026年，Model Context Protocol（MCP）已經成為AI工具呼叫的開放標準——寫一次MCP Server，所有主流AI平台都能呼叫。

背景知識

MCP協定架構

MCP（Model Context Protocol）是Anthropic在2024年底發佈的開放協定，定義了AI應用（Host）與工具服務（Server）之間的標準通訊方式。

概念 說明 類比
Host AI應用（Claude Desktop、ChatGPT等） 瀏覽器
Client Host內部的MCP客戶端 HTTP Client
Server 提供工具/資源/提示的服務 HTTP Server
Transport 通訊方式（stdio/SSE/Streamable HTTP） TCP/HTTP

MCP核心能力

能力 說明 範例
Tools 可被AI呼叫的函式 查詢資料庫、傳送電子郵件
Resources 可被AI讀取的資料 檔案內容、資料庫記錄
Prompts 預先定義的提示範本 程式碼審查範本
Sampling AI向Host請求生成 需要AI判斷的中間步驟

傳輸方式對比

傳輸 適用場景 優點 缺點
stdio 本機桌面應用 簡單、安全 僅限本機行程
SSE 遠端Web服務 跨網路 長連線維護
Streamable HTTP 遠端Web服務（推薦） 無狀態、CDN友善 需MCP 2025-03規範

問題分析

為什麼需要MCP？

  1. 協定碎片化：每個AI平台的工具呼叫格式不同，適配成本高
  2. 工具復用難：為Claude寫的工具無法在ChatGPT中使用
  3. 安全風險：直接暴露API端點給AI，缺乏權限控制
  4. 狀態管理混亂：AI與工具之間的上下文傳遞沒有標準

MCP透過統一協定解決這些問題：一個MCP Server可以被任何支援MCP的AI Host呼叫。

分步實作

步驟1：安裝MCP Python SDK

pip install mcp[cli]

步驟2：建立第一個MCP Server

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("toolsku-tools")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="calculate",
            description="執行數學表達式計算",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "數學表達式，如 '2 + 3 * 4'",
                    }
                },
                "required": ["expression"],
            },
        ),
        Tool(
            name="format_json",
            description="格式化JSON字串",
            inputSchema={
                "type": "object",
                "properties": {
                    "json_str": {
                        "type": "string",
                        "description": "待格式化的JSON字串",
                    },
                    "indent": {
                        "type": "integer",
                        "description": "縮排空格數，預設2",
                        "default": 2,
                    },
                },
                "required": ["json_str"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "calculate":
        try:
            result = eval(arguments["expression"], {"__builtins__": {}}, {})
            return [TextContent(type="text", text=f"計算結果: {result}")]
        except Exception as e:
            return [TextContent(type="text", text=f"計算錯誤: {str(e)}")]

    elif name == "format_json":
        import json
        try:
            parsed = json.loads(arguments["json_str"])
            indent = arguments.get("indent", 2)
            formatted = json.dumps(parsed, indent=indent, ensure_ascii=False)
            return [TextContent(type="text", text=formatted)]
        except json.JSONDecodeError as e:
            return [TextContent(type="text", text=f"JSON解析錯誤: {str(e)}")]

    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

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

步驟3：設定Claude Desktop連線

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%\Claude\claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "toolsku-tools": {
      "command": "python",
      "args": ["server.py"],
      "cwd": "E:\\project\\toolsku\\mcp-server"
    }
  }
}

步驟4：SSE傳輸模式（遠端部署）

# sse_server.py
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.routing import Mount, Route

app = Server("toolsku-remote-tools")
sse = SseServerTransport("/messages/")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="搜尋技術文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜尋關鍵字"},
                    "category": {"type": "string", "description": "文件分類"},
                    "limit": {"type": "integer", "description": "回傳數量", "default": 5},
                },
                "required": ["query"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_docs":
        results = await search_documents(
            query=arguments["query"],
            category=arguments.get("category"),
            limit=arguments.get("limit", 5),
        )
        import json
        return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))]
    raise ValueError(f"Unknown tool: {name}")

async def search_documents(query: str, category: str | None, limit: int) -> list[dict]:
    return [
        {"title": f"文件: {query}", "url": f"https://docs.example.com/{query}", "score": 0.95},
    ][:limit]

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8080)

步驟5：與ChatGPT整合

# chatgpt_adapter.py
import httpx
from openai import OpenAI

MCP_SERVER_URL = "http://localhost:8080"

async def discover_mcp_tools() -> list[dict]:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MCP_SERVER_URL}/messages/",
            json={
                "jsonrpc": "2.0",
                "method": "tools/list",
                "id": 1,
            },
        )
        result = response.json()
        return result.get("result", {}).get("tools", [])

def mcp_tools_to_openai_tools(mcp_tools: list[dict]) -> list[dict]:
    openai_tools = []
    for tool in mcp_tools:
        openai_tools.append({
            "type": "function",
            "function": {
                "name": tool["name"],
                "description": tool["description"],
                "parameters": tool["inputSchema"],
            },
        })
    return openai_tools

async def call_mcp_tool(name: str, arguments: dict) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MCP_SERVER_URL}/messages/",
            json={
                "jsonrpc": "2.0",
                "method": "tools/call",
                "params": {"name": name, "arguments": arguments},
                "id": 2,
            },
        )
        result = response.json()
        contents = result.get("result", {}).get("content", [])
        return "\n".join(c.get("text", "") for c in contents)

async def chat_with_tools(user_message: str) -> str:
    mcp_tools = await discover_mcp_tools()
    openai_tools = mcp_tools_to_openai_tools(mcp_tools)

    client = OpenAI()
    messages = [{"role": "user", "content": user_message}]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=openai_tools,
    )

    message = response.choices[0].message

    if message.tool_calls:
        for tool_call in message.tool_calls:
            tool_result = await call_mcp_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments),
            )
            messages.append(message)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": tool_result,
            })

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=openai_tools,
        )
        return response.choices[0].message.content

    return message.content

完整程式碼：生產級MCP Server

# production_server.py
import os
import json
import logging
from datetime import datetime
from typing import Any

from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent, ImageContent
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("toolsku-mcp")

app = Server("toolsku-production")

API_KEY = os.environ.get("MCP_API_KEY", "dev-key")

def validate_api_key(headers: dict) -> bool:
    auth = headers.get("authorization", "")
    return auth == f"Bearer {API_KEY}"

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="base64_encode",
            description="Base64編碼字串",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "待編碼的文字"},
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="base64_decode",
            description="Base64解碼字串",
            inputSchema={
                "type": "object",
                "properties": {
                    "encoded": {"type": "string", "description": "Base64編碼的字串"},
                },
                "required": ["encoded"],
            },
        ),
        Tool(
            name="json_format",
            description="格式化JSON字串",
            inputSchema={
                "type": "object",
                "properties": {
                    "json_str": {"type": "string", "description": "JSON字串"},
                    "indent": {"type": "integer", "description": "縮排", "default": 2},
                    "sort_keys": {"type": "boolean", "description": "是否排序鍵", "default": False},
                },
                "required": ["json_str"],
            },
        ),
        Tool(
            name="hash_compute",
            description="計算文字的雜湊值",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "待計算的文字"},
                    "algorithm": {
                        "type": "string",
                        "enum": ["md5", "sha1", "sha256", "sha512"],
                        "description": "雜湊演算法",
                        "default": "sha256",
                    },
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="timestamp_convert",
            description="時間戳與日期格式互轉",
            inputSchema={
                "type": "object",
                "properties": {
                    "value": {"type": "string", "description": "時間戳或日期字串"},
                    "format": {"type": "string", "description": "目標格式", "default": "iso"},
                },
                "required": ["value"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    logger.info(f"Tool called: {name} with args: {arguments}")

    try:
        if name == "base64_encode":
            import base64
            result = base64.b64encode(arguments["text"].encode()).decode()
            return [TextContent(type="text", text=result)]

        elif name == "base64_decode":
            import base64
            result = base64.b64decode(arguments["encoded"]).decode()
            return [TextContent(type="text", text=result)]

        elif name == "json_format":
            parsed = json.loads(arguments["json_str"])
            indent = arguments.get("indent", 2)
            sort_keys = arguments.get("sort_keys", False)
            formatted = json.dumps(parsed, indent=indent, sort_keys=sort_keys, ensure_ascii=False)
            return [TextContent(type="text", text=formatted)]

        elif name == "hash_compute":
            import hashlib
            algorithm = arguments.get("algorithm", "sha256")
            h = hashlib.new(algorithm)
            h.update(arguments["text"].encode())
            return [TextContent(type="text", text=h.hexdigest())]

        elif name == "timestamp_convert":
            value = arguments["value"]
            fmt = arguments.get("format", "iso")
            try:
                ts = float(value)
                dt = datetime.fromtimestamp(ts)
            except ValueError:
                dt = datetime.fromisoformat(value)
                ts = dt.timestamp()

            if fmt == "iso":
                result = dt.isoformat()
            elif fmt == "timestamp":
                result = str(ts)
            else:
                result = dt.strftime(fmt)

            return [TextContent(type="text", text=result)]

        raise ValueError(f"Unknown tool: {name}")

    except Exception as e:
        logger.error(f"Tool error: {name} - {str(e)}")
        return [TextContent(type="text", text=f"錯誤: {str(e)}")]

sse = SseServerTransport("/messages/")

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),
]

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
    middleware=middleware,
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8080)

避坑指南

坑1：Claude Desktop無法發現MCP Server

現象：設定了claude_desktop_config.json但Claude Desktop看不到工具。

解決：檢查JSON格式是否正確（注意逗號和引號），確認command路徑正確（使用絕對路徑），檢視Claude Desktop日誌（Help > Toggle Developer Tools）。

坑2：SSE連線頻繁斷開

現象：遠端MCP Server的SSE連線每隔30秒斷開。

解決：確保服務端實作了心跳機制（每15秒傳送keep-alive），設定反向代理的超時時間大於60秒。Nginx需要proxy_read_timeout 300s。

坑3：工具參數校驗不嚴格導致執行時錯誤

現象：AI傳入了錯誤型別的參數（如字串傳給了需要整數的參數）。

解決：在inputSchema中嚴格定義型別和約束（type、minimum、enum等），在call_tool函式中新增參數校驗和型別轉換。

坑4：stdio模式下Python環境不一致

現象：Claude Desktop啟動MCP Server時找不到已安裝的套件。

解決：在claude_desktop_config.json中使用Python直譯器的絕對路徑，或使用uv run指定虛擬環境：

{
  "mcpServers": {
    "toolsku-tools": {
      "command": "uv",
      "args": ["run", "--directory", "E:\\project\\toolsku\\mcp-server", "server.py"]
    }
  }
}

坑5：並行請求導致狀態混亂

現象：多個AI同時呼叫同一個MCP Server，共享狀態被覆寫。

解決：MCP Server應設計為無狀態，所有狀態透過參數傳遞。如果必須有狀態，使用請求級別的上下文隔離，避免全域變數。

報錯排查

序號 報錯資訊 原因 解決方法
1 MCP server not found in config 設定檔路徑或格式錯誤 檢查JSON格式，使用絕對路徑
2 Connection refused on stdio Python行程啟動失敗 檢查command路徑，檢視stderr日誌
3 SSE connection timeout 服務端未回應或網路問題 檢查服務端狀態，增加超時設定
4 Tool not found: xxx 工具名拼寫錯誤 檢查list_tools和call_tool的名稱一致
5 JSON-RPC error: -32600 請求格式不符合JSON-RPC規範 檢查請求body格式
6 Permission denied 工具執行需要權限 檢查檔案/網路權限，使用sudo謹慎處理
7 ModuleNotFoundError Python依賴未安裝 在MCP設定中指定正確的Python環境
8 413 Payload Too Large 請求body過大 限制工具參數大小，使用分頁
9 CORS error SSE跨域請求被攔截 設定CORS中介軟體
10 Server initialization failed MCP交握失敗 檢查initialize回應格式和capabilities

進階最佳化

1. 資源（Resources）暴露

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="toolsku:///docs/api-reference",
            name="API參考文件",
            mimeType="text/markdown",
        ),
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "toolsku:///docs/api-reference":
        return "# API參考\n\n## 工具列表\n..."
    raise ValueError(f"Unknown resource: {uri}")

2. 提示範本（Prompts）

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="code-review",
            description="程式碼審查提示範本",
            arguments=[
                PromptArgument(name="language", description="程式語言", required=True),
                PromptArgument(name="code", description="待審查的程式碼", required=True),
            ],
        ),
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
    if name == "code-review":
        return [
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"請審查以下{arguments['language']}程式碼，關注安全性、效能和可維護性：\n\n```{arguments['language']}\n{arguments['code']}\n```",
                ),
            ),
        ]
    raise ValueError(f"Unknown prompt: {name}")

3. Streamable HTTP傳輸（MCP 2025-03）

from mcp.server.streamable_http import StreamableHTTPTransport

http_transport = StreamableHTTPTransport("/mcp")

async def handle_http(request):
    await http_transport.handle_request(request, app)

4. 認證與限流

from starlette.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path in ["/sse", "/messages/"]:
            auth = request.headers.get("authorization", "")
            if auth != f"Bearer {API_KEY}":
                from starlette.responses import JSONResponse
                return JSONResponse({"error": "unauthorized"}, status_code=401)
        return await call_next(request)

對比分析

維度 MCP OpenAI Function Calling LangChain Tools LlamaIndex Tools
協定標準 開放標準 平台私有 框架私有 框架私有
跨平台 所有支援MCP的Host 僅OpenAI 僅LangChain 僅LlamaIndex
傳輸方式 stdio/SSE/HTTP HTTP API Python函式 Python函式
工具發現 動態list_tools 靜態定義 靜態定義 靜態定義
資源暴露 原生支援 不支援 不支援 不支援
安全模型 權限控制 API Key 無標準 無標準
部署方式 本機/遠端 遠端 行程內 行程內
生態 快速增長 成熟 成熟 成熟

總結展望

總結：MCP在2026年已成為AI工具呼叫的開放標準。Python MCP SDK讓開發一個MCP Server變得極其簡單——定義工具、註冊處理器、選擇傳輸方式。核心優勢是「一次開發，多平台呼叫」：同一個MCP Server可以被Claude Desktop、ChatGPT、Cursor等任何支援MCP的AI應用使用。建議從簡單的工具服務入手（如JSON格式化、Base64編解碼），逐步新增資源和提示範本能力，最終建構完整的AI工具生態。

線上工具推薦

本站提供瀏覽器本地工具，免註冊即可試用 →

#Python#MCP#Model Context Protocol#AI工具#SSE#stdio#Agent#工具服务