介紹

MCP 使用 JSON-RPC 2.0 作為其傳輸格式。傳輸層負責將 MCP 協議消息轉換為 JSON-RPC 格式進行傳輸，并將接收到的 JSON-RPC 消息轉換回 MCP 協議消息。其中SSE被廢棄了（Server-Sent Events (SSE) - Deprecated）

SSE?as?a?standalone?transport?is?deprecated?as?of?protocol?version?2024-11-05.?It?has?been?replaced?by?Streamable?HTTP,?which?incorporates?SSE?as?an?optional?streaming?mechanism.?For?backwards?compatibility?information,?see?the?backwards?compatibility?section?below.

自協議版本 2024-11-05 起，SSE 作為獨立傳輸已被棄用。它已被流式 HTTP 取代，后者將 SSE 作為可選的流式機制。有關向后兼容性信息，請參閱下方的向后兼容性部分。

MCP官網：Transports - Model Context Protocol

實戰

1.編寫一個py代碼

from __future__ import annotations
import argparse
import asyncio
import json
from typing import Any, AsyncIterator
import httpx
from fastapi import FastAPI, Request, Response, status
from fastapi.responses import StreamingResponse
# ---------------------------------------------------------------------------
# Server constants
# ---------------------------------------------------------------------------
SERVER_NAME = "WeatherServer"
SERVER_VERSION = "1.0.0"
PROTOCOL_VERSION = "2024-11-05" # Cherry Studio current
# ---------------------------------------------------------------------------
# Weather helpers
# ---------------------------------------------------------------------------
OPENWEATHER_URL = "https://api.openweathermap.org/data/2.5/weather"
API_KEY: str | None = None
USER_AGENT = "weather-app/1.0"
async def fetch_weather(city: str) -> dict[str, Any]:if not API_KEY:return {"error": "API_KEY 未設置，請提供有效的 OpenWeather API Key。"}params = {"q": city, "appid": API_KEY, "units": "metric", "lang": "zh_cn"}headers = {"User-Agent": USER_AGENT}async with httpx.AsyncClient(timeout=30.0) as client:try:r = await client.get(OPENWEATHER_URL, params=params, headers=headers)r.raise_for_status()return r.json()except httpx.HTTPStatusError as exc:return {"error": f"HTTP 錯誤: {exc.response.status_code} - {exc.response.text}"}except httpx.RequestError as exc:return {"error": f"網絡請求失敗: {exc}"}except json.JSONDecodeError:return {"error": "OpenWeather 返回的數據不是有效的 JSON"}def format_weather(data: dict[str, Any]) -> str:if "error" in data:return data["error"]city = data.get("name", "未知")country = data.get("sys", {}).get("country", "未知")temp = data.get("main", {}).get("temp", "N/A")humidity = data.get("main", {}).get("humidity", "N/A")wind = data.get("wind", {}).get("speed", "N/A")desc = data.get("weather", [{}])[0].get("description", "未知")return (f"🌍 {city}, {country}\n"f"🌡 溫度: {temp}°C\n"f"💧 濕度: {humidity}%\n"f"🌬 風速: {wind} m/s\n"f"🌤 天氣: {desc}")async def stream_weather(city: str, req_id: int | str) -> AsyncIterator[bytes]:# 進度提示（不會出錯的部分放在try外面）yield json.dumps({"jsonrpc": "2.0","id": req_id,"stream": f"查詢{city}天氣中…"}).encode() + b"\n"try:await asyncio.sleep(0.3)data = await fetch_weather(city)  # 主要可能出錯點if "error" in data:yield json.dumps({"jsonrpc": "2.0","id": req_id,"error": {"code": -32002,  # 保留業務錯誤碼"message": data["error"]}}).encode() + b"\n"return# 正常結果（json.dumps理論上不會出錯）yield json.dumps({"jsonrpc": "2.0","id": req_id,"result": {"content": [{"type": "text", "text": format_weather(data)}],"isError": False}}).encode() + b"\n"except Exception as exc:# 系統級錯誤單獨處理yield json.dumps({"jsonrpc": "2.0","id": req_id,"error": {"code": -32003,  # 保留系統錯誤碼"message": f"服務器內部錯誤: {str(exc)}"}}).encode() + b"\n"# ---------------------------------------------------------------------------# FastAPI app# ---------------------------------------------------------------------------
app = FastAPI(title="WeatherServer HTTP-Stream v8")
TOOLS_REGISTRY = {"tools": [{"name": "get_weather","description": "用于進行天氣信息查詢的函數，輸入城市英文名稱，即可獲得當前城市天氣信息。","inputSchema": {"type": "object","properties": {"city": {"type": "string","description": "City name, e.g. 'Hangzhou'"}},"required": ["city"]}}],"nextCursor": "qinqing"
}
@app.get("/mcp")
async def mcp_initialize_via_get():
# GET 請求也執行了 initialize 方法return {"jsonrpc": "2.0","id": 0,"result": {"protocolVersion": PROTOCOL_VERSION,"capabilities": {"streaming": True,"tools": {"listChanged": True}},"serverInfo": {"name": SERVER_NAME,"version": SERVER_VERSION},"instructions": "Use the get_weather tool to fetch weather by cityname."}
}
@app.post("/mcp")
async def mcp_endpoint(request: Request):try:body = await request.json()# ? 打印客戶端請求內容print("💡 收到請求:", json.dumps(body, ensure_ascii=False, indent=2))except Exception:return {"jsonrpc": "2.0", "id": None, "error": {"code": -32700,"message": "Parse error"}}req_id = body.get("id", 1)method = body.get("method")# ? 打印當前方法類型print(f"🔧 方法: {method}")# 0) Ignore initialized notification (no response required)if method == "notifications/initialized":return Response(status_code=status.HTTP_204_NO_CONTENT)# 1) Activation probe (no method)if method is None:return {"jsonrpc": "2.0", "id": req_id, "result": {"status": "MCP serveronline."}}# 2) initializeif method == "initialize":return {"jsonrpc": "2.0","id": req_id,"result": {"protocolVersion": PROTOCOL_VERSION,"capabilities": {"streaming": True,"tools": {"listChanged": True}},"serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},"instructions": "Use the get_weather tool to fetch weather bycity name."}}# 3) tools/listif method == "tools/list":print(json.dumps(TOOLS_REGISTRY, indent=2, ensure_ascii=False))return {"jsonrpc": "2.0", "id": req_id, "result": TOOLS_REGISTRY}# 4) tools/callif method == "tools/call":params = body.get("params", {})tool_name = params.get("name")args = params.get("arguments", {})if tool_name != "get_weather":return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32602,"message": "Unknown tool"}}city = args.get("city")if not city:return {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32602,"message": "Missing city"}}# return StreamingResponse(stream_weather(city, req_id),media_type="application/json")data = await fetch_weather(city)return {"jsonrpc": "2.0","id": req_id,"result": {"content": [{"type": "text", "text": format_weather(data)}],"isError": False}}# 5) unknown methodreturn {"jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message":"Method not found"}}
# ---------------------------------------------------------------------------
# Runner
# ---------------------------------------------------------------------------
def main() -> None:parser = argparse.ArgumentParser(description="Weather MCP HTTP-Stream v8")parser.add_argument("--api_key", required=False,default='xxxx')parser.add_argument("--host", default="127.0.0.1")parser.add_argument("--port", type=int, default=8000)args = parser.parse_args()global API_KEYAPI_KEY = args.api_keyimport uvicornuvicorn.run(app, host=args.host, port=args.port, log_level="info")
if __name__ == "__main__":# npx -y @modelcontextprotocol/inspector uv run weather2.pymain()

其中api_key 需要自己申請即可，可以免費調用大概1000次（測試和學習基本夠用了）?

2. 使用mcp的檢查器進行測試

npx -y @modelcontextprotocol/inspector uv run xxxx.py

參考官網：Inspector - Model Context Protocol

3. 將py進行啟動運行

4. 檢查器啟動成功之后進行測試

簡單的例子Streamable HTTP就成功了 ~