MCP協議實戰：用Model Context Protocol構建AI Agent工具鏈

AI与大数据

摘要

  • MCP（Model Context Protocol）是Anthropic提出的AI Agent工具呼叫標準協議，2026年已成為事實上的Agent工具鏈協議
  • MCP透過JSON-RPC 2.0實現工具註冊、發現和呼叫，解決了AI Agent「工具孤島」問題
  • 本文從MCP協議原理到Server開發，從工具註冊到多Agent編排，全鏈路完整程式碼
  • MCP支援SSE和Stdio兩種傳輸模式，SSE適合遠端部署，Stdio適合本機開發
  • 附贈MCP Server生產部署方案與工具鏈安全稽核checklist

目錄

為什麼AI Agent需要MCP協議

2026年之前，AI Agent呼叫工具的方式是「各自為政」——OpenAI用Function Calling，LangChain用Tool抽象，AutoGPT用自訂外掛程式。每個框架都定義了自己的工具介面，工具無法跨框架複用，Agent無法跨平台協作。

` ┌──────────────────────────────────────────────────────────────┐ │ AI Agent工具呼叫的「巴別塔」 │ │ │ │ OpenAI Agent ──→ Function Calling (JSON Schema) │ │ LangChain ──→ Tool Abstraction (Python Class) │ │ AutoGPT ──→ Plugin System (YAML Config) │ │ Dify ──→ Tool Node (API Config) │ │ Coze ──→ Plugin Market (Proprietary) │ │ │ │ ❌ 工具無法複用 ❌ 介面不統一 ❌ 安全無保障 ❌ 發現無標準 │ └──────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────┐ │ MCP協議：AI Agent的「USB-C」 │ │ │ │ 任何Agent ←──→ MCP協議 ←──→ 任何工具 │ │ │ │ ┌────────┐ MCP ┌──────────┐ MCP ┌──────────┐ │ │ │ Claude │←──────→│ 搜尋工具 │←──────→│ GPT-4 │ │ │ │ Agent │ │ 資料庫 │ │ Agent │ │ │ └────────┘ │ API閘道 │ └──────────┘ │ │ └──────────┘ │ │ │ │ ✅ 工具複用 ✅ 介面統一 ✅ 安全沙箱 ✅ 自動發現 │ └──────────────────────────────────────────────────────────────┘ `

MCP vs Function Calling vs LangChain Tool

維度 MCP OpenAI Function Calling LangChain Tool
協議標準 開放標準(Anthropic) 廠商私有 框架私有
工具發現 自動發現(capabilities) 手動註冊 手動註冊
傳輸協議 JSON-RPC 2.0 (SSE/Stdio) HTTP API Python函式呼叫
安全沙箱 ✅ 權限控制 ❌ ❌
跨Agent複用 ✅ 任何MCP Client ❌ 僅OpenAI ❌ 僅LangChain
串流輸出 ✅ SSE ✅ SSE ⚠️ 部分支援
多工具編排 ✅ 原生支援 ⚠️ 需手動 ✅ Chain抽象

參考：Model Context Protocol Specification

MCP協議核心機制

三大角色

角色 說明 類比
MCP Host 發起連線的AI應用（如Claude Desktop、IDE外掛程式） USB主機
MCP Client 與MCP Server建立連線的協議客戶端，嵌入在Host中 USB控制器
MCP Server 提供工具、資源、提示詞的伺服器端 USB裝置

協議互動流程

┌──────────────────────────────────────────────────────────────┐ │ MCP協議互動流程 │ │ │ │ MCP Host (Claude Desktop) │ │ │ │ │ │ 1. initialize │ │ ├──→ MCP Client ──→ MCP Server (搜尋工具) │ │ │ │ │ │ │ 2. capabilities回應 │ │ │ │←─────────────────────────────────────────┘ │ │ │ │ │ │ 3. tools/list │ │ ├──→ MCP Client ──→ MCP Server │ │ │ │ │ │ │ 4. 工具清單回應 │ │ │ │←─────────────────────────────────────────┘ │ │ │ │ │ │ 5. tools/call (搜尋"K8s GPU排程") │ │ ├──→ MCP Client ──→ MCP Server │ │ │ │ │ │ │ 6. 工具執行結果 │ │ │ │←─────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────────────┘

MCP能力類型

能力 說明 範例
Tools 可被Agent呼叫的函式 搜尋、資料庫查詢、API呼叫
Resources 可被Agent讀取的資料 檔案、資料庫記錄、設定
Prompts 可被Agent使用的提示詞範本 程式碼審查範本、翻譯範本
Sampling Server向Agent請求LLM補全 多步推理、內容生成

MCP Server開發：5步構建自訂工具服務

第1步：專案初始化

ash mkdir mcp-search-server && cd mcp-search-server pip install mcp fastapi httpx

第2步：定義工具Schema

`python from mcp.server import Server from mcp.types import Tool, TextContent import httpx import json

server = Server("search-tools")

@server.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="web_search", description="搜尋網際網路取得最新資訊。當需要查詢即時資料、新聞、技術文件時使用此工具。", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "搜尋關鍵字" }, "max_results": { "type": "integer", "description": "最大回傳結果數", "default": 5 }, "search_depth": { "type": "string", "enum": ["basic", "advanced"], "description": "搜尋深度：basic快速搜尋，advanced深度搜尋", "default": "basic" } }, "required": ["query"] } ), Tool( name="code_search", description="在GitHub上搜尋程式碼儲存庫和程式碼片段。當需要查詢開源實作、API用法範例時使用。", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "程式碼搜尋關鍵字" }, "language": { "type": "string", "description": "程式語言篩選", "enum": ["python", "go", "rust", "typescript", "java"] } }, "required": ["query"] } ), Tool( name="api_tester", description="傳送HTTP請求測試API端點。當需要驗證API可用性、取得API回應資料時使用。", inputSchema={ "type": "object", "properties": { "url": { "type": "string", "description": "API端點URL" }, "method": { "type": "string", "enum": ["GET", "POST", "PUT", "DELETE"], "default": "GET" }, "headers": { "type": "object", "description": "請求標頭" }, "body": { "type": "object", "description": "請求主體(JSON)" } }, "required": ["url"] } ), ] `

第3步：實作工具邏輯

`python @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: if name == "web_search": return await handle_web_search(arguments) elif name == "code_search": return await handle_code_search(arguments) elif name == "api_tester": return await handle_api_tester(arguments) else: raise ValueError(f"Unknown tool: {name}")

async def handle_web_search(args: dict) -> list[TextContent]: query = args["query"] max_results = args.get("max_results", 5) search_depth = args.get("search_depth", "basic")

async with httpx.AsyncClient(timeout=30) as client:
    response = await client.post(
        "https://api.search.brave.com/res/v1/web/search",
        headers={"X-Subscription-Token": BRAVE_API_KEY},
        params={
            "q": query,
            "count": max_results,
            "search_depth": search_depth,
        }
    )
    data = response.json()

results = []
for item in data.get("web", {}).get("results", [])[:max_results]:
    results.append({
        "title": item.get("title", ""),
        "url": item.get("url", ""),
        "description": item.get("description", ""),
    })

return [TextContent(
    type="text",
    text=json.dumps(results, ensure_ascii=False, indent=2)
)]

async def handle_code_search(args: dict) -> list[TextContent]: query = args["query"] language = args.get("language", "")

async with httpx.AsyncClient(timeout=30) as client:
    response = await client.get(
        "https://api.github.com/search/code",
        headers={"Authorization": f"token {GITHUB_TOKEN}"},
        params={
            "q": f"{query} language:{language}" if language else query,
            "per_page": 5,
        }
    )
    data = response.json()

results = []
for item in data.get("items", [])[:5]:
    results.append({
        "name": item.get("name", ""),
        "path": item.get("path", ""),
        "repository": item.get("repository", {}).get("full_name", ""),
        "html_url": item.get("html_url", ""),
    })

return [TextContent(
    type="text",
    text=json.dumps(results, ensure_ascii=False, indent=2)
)]

async def handle_api_tester(args: dict) -> list[TextContent]: url = args["url"] method = args.get("method", "GET") headers = args.get("headers", {}) body = args.get("body")

async with httpx.AsyncClient(timeout=30) as client:
    response = await client.request(
        method=method,
        url=url,
        headers=headers,
        json=body,
    )

return [TextContent(
    type="text",
    text=json.dumps({
        "status_code": response.status_code,
        "headers": dict(response.headers),
        "body": response.json() if "json" in response.headers.get("content-type", "") else response.text[:2000],
    }, ensure_ascii=False, indent=2)
)]

`

第4步：設定SSE傳輸

`python from mcp.server.sse import SseServerTransport from starlette.applications import Starlette from starlette.routing import Route

sse = SseServerTransport("/messages")

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

async def handle_messages(request): await sse.handle_post_message(request)

app = Starlette( routes=[ Route("/sse", endpoint=handle_sse), Route("/messages", endpoint=handle_messages, methods=["POST"]), ] )

if name == "main": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8080) `

第5步：Docker部署

`dockerfile FROM python:3.11-slim

WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . .

EXPOSE 8080 HEALTHCHECK --interval=30s --timeout=5s --retries=3
CMD curl -f http://localhost:8080/sse || exit 1

CMD ["python", "server.py"] `

MCP Client整合：讓大模型呼叫工具

Python MCP Client

`python from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from openai import OpenAI import json

server_params = StdioServerParameters( command="python", args=["server.py"], env={"BRAVE_API_KEY": "...", "GITHUB_TOKEN": "..."} )

async def run_agent(user_query: str): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize()

        tools_result = await session.list_tools()
        available_tools = [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.inputSchema,
                }
            }
            for tool in tools_result.tools
        ]

        client = OpenAI(base_url="http://localhost:8000/v1", api_key="not-needed")

        messages = [{"role": "user", "content": user_query}]

        response = client.chat.completions.create(
            model="Qwen/Qwen2.5-7B-Instruct",
            messages=messages,
            tools=available_tools,
            tool_choice="auto",
        )

        message = response.choices[0].message

        if message.tool_calls:
            for tool_call in message.tool_calls:
                result = await session.call_tool(
                    tool_call.function.name,
                    arguments=json.loads(tool_call.function.arguments),
                )

                messages.append(message)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result[0].text,
                })

            final_response = client.chat.completions.create(
                model="Qwen/Qwen2.5-7B-Instruct",
                messages=messages,
            )
            return final_response.choices[0].message.content

        return message.content

`

工具呼叫流程

┌──────────────────────────────────────────────────────────────┐ │ MCP Agent工具呼叫完整流程 │ │ │ │ 使用者：「幫我搜尋K8s GPU排程的最新方案並測試相關API」 │ │ │ │ │ ▼ │ │ LLM分析意圖 → 需要呼叫2個工具： │ │ 1. web_search("K8s GPU排程 2026") │ │ 2. api_tester("https://kubernetes.io/docs/concepts/") │ │ │ │ │ ├──→ MCP Client ──→ web_search ──→ 回傳搜尋結果 │ │ │ │ │ ├──→ MCP Client ──→ api_tester ──→ 回傳API回應 │ │ │ │ │ ▼ │ │ LLM整合結果 → 產生最終回答 │ └──────────────────────────────────────────────────────────────┘

多Agent編排：MCP工具鏈組合

多Server編排架構

`python from mcp import ClientSession from mcp.client.sse import sse_client import asyncio

class MCPToolOrchestrator: def init(self): self.servers = {} self.sessions = {}

async def register_server(self, name: str, url: str):
    self.servers[name] = url

async def connect_all(self):
    for name, url in self.servers.items():
        read, write = await sse_client(url)
        session = ClientSession(read, write)
        await session.initialize()
        self.sessions[name] = session

async def discover_all_tools(self) -> list[dict]:
    all_tools = []
    for name, session in self.sessions.items():
        tools = await session.list_tools()
        for tool in tools.tools:
            all_tools.append({
                "server": name,
                "name": tool.name,
                "description": tool.description,
                "inputSchema": tool.inputSchema,
            })
    return all_tools

async def call_tool(self, tool_name: str, arguments: dict) -> str:
    for name, session in self.sessions.items():
        tools = await session.list_tools()
        if any(t.name == tool_name for t in tools.tools):
            result = await session.call_tool(tool_name, arguments)
            return result[0].text
    raise ValueError(f"Tool {tool_name} not found in any server")

async def close_all(self):
    for session in self.sessions.values():
        await session.close()

orchestrator = MCPToolOrchestrator() await orchestrator.register_server("search", "http://search-mcp:8080/sse") await orchestrator.register_server("database", "http://db-mcp:8081/sse") await orchestrator.register_server("code", "http://code-mcp:8082/sse") await orchestrator.connect_all() `

K8s部署MCP Server叢集

`yaml apiVersion: apps/v1 kind: Deployment metadata: name: mcp-search-server namespace: ai-agent spec: replicas: 2 selector: matchLabels: app: mcp-search-server template: metadata: labels: app: mcp-search-server spec: containers: - name: mcp-server image: myregistry/mcp-search-server:v1.0 ports: - containerPort: 8080 resources: requests: cpu: "1" memory: 512Mi limits: cpu: "2" memory: 1Gi env: - name: BRAVE_API_KEY valueFrom: secretKeyRef: name: mcp-secrets key: brave-api-key livenessProbe: httpGet: path: /sse port: 8080 initialDelaySeconds: 10 periodSeconds: 30

apiVersion: v1 kind: Service metadata: name: mcp-search-svc namespace: ai-agent spec: selector: app: mcp-search-server ports: - port: 8080 targetPort: 8080 `

生產部署與安全稽核

MCP安全模型

安全層 機制 說明
傳輸層 TLS + SSE 加密傳輸，防止中間人攻擊
認證層 OAuth 2.0 / API Key Server驗證Client身分
授權層 工具權限白名單 限制可呼叫的工具範圍
資料層 輸入校驗 + 輸出過濾 防止注入攻擊和資料外洩
稽核層 呼叫日誌 + 計量 全鏈路追蹤和費用控制

安全稽核Checklist

檢查項目 風險等級 檢查方法
工具輸入參數校驗 高 Schema驗證 + 型別檢查
工具輸出資料去識別化 高 正規表示式過濾敏感資訊
API Key不硬編碼 高 環境變數/Secret管理
SSE連線逾時控制 中 心跳 + 逾時斷開
工具呼叫頻率限制 中 Rate Limiting
呼叫日誌完整性 中 結構化日誌 + 稽核追蹤
錯誤訊息不洩露內部狀態 低 統一錯誤格式

MCP Server效能基準

指標 單實例 3實例叢集
工具呼叫QPS 500 1400
平均延遲(P50) 45ms 50ms
P99延遲 180ms 200ms
SSE連線數 200 600
記憶體佔用 256MB 256MB×3

總結與延伸閱讀

MCP協議是AI Agent工具鏈的「USB-C」——一個開放標準解決了工具孤島、介面不統一、安全無保障三大問題。透過MCP Server開發、Client整合和多Agent編排，可以快速構建生產級AI Agent工具鏈。

開發要點回顧：

  1. MCP透過JSON-RPC 2.0實現工具註冊、發現和呼叫
  2. SSE模式適合遠端部署，Stdio模式適合本機開發
  3. 工具Schema定義是MCP的核心，需詳細描述參數和語意
  4. 多Agent編排透過MCPToolOrchestrator實現跨Server工具呼叫
  5. 生產部署需關注安全稽核：輸入校驗、輸出去識別化、頻率限制

相關閱讀：

權威參考：

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

#MCP协议#AI Agent工具调用#Model Context Protocol#大模型工具使用#Agent开发框架#2026