High Altitude Oolong: MCP Weather Server：SSE Transport

SSE (Server-Sent Events) transport，這是一個基於 HTTP 的傳輸方式，讓 MCP server 可以作為獨立的網路服務運行。

SSE Transport

SSE transport 有以下特點：

透過 HTTP 通訊，而非標準輸入/輸出
server 作為獨立的 HTTP 服務運行
client 透過網路連線到 server
使用 Server-Sent Events 實作雙向通訊
可同時服務多個 client

架構

SSE server 的核心檔案位於 src/weather_server/sse_server.py，包含以下組件：

MCP server 實例：處理工具註冊和調用
SSE transport：管理 HTTP 連線
Starlette application：提供 HTTP endpoints
Uvicorn server：運行 ASGI 應用程式

實作

1. 建立 MCP Server

建立 MCP server 並註冊：

from mcp.server import Server
from mcp.server.sse import SseServerTransport

mcp_server = Server("weather-server-sse")

@mcp_server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_weather",
            description="Get current weather information for a city.",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "Name of the city to get weather for",
                    }
                },
                "required": ["city"],
            },
        )
    ]

2. 設定 SSE Transport

建立 SSE transport ，指定訊息處理的路徑：

sse = SseServerTransport("/messages")

3. 實作 HTTP Endpoints

SSE transport 需要兩個 HTTP endpoints：

GET /sse - 建立 SSE 連線

async def handle_sse(request: Request):
    logger.info(f"New SSE connection from {request.client.host}")

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

此 endpoint 接收來自 client 的連線請求，並建立持久的 SSE 連線。

POST /messages - 接收訊息

async def handle_messages(request: Request):
    await sse.handle_post_message(request.scope, request.receive, request._send)

此 endpoint 處理 client 發送的 POST 請求，實作雙向通訊的「上行」部分。

4. 建立 Starlette Application

from starlette.applications import Starlette
from starlette.routing import Route

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

5. 啟動 Server

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(
        app,
        host="localhost",
        port=3000,
        log_level="info"
    )

執行 SSE Server

1. 直接執行

python -m weather_server.sse_server

server 將在 http://localhost:3000 啟動。

2. 使用 uv run（推薦）

uv run weather-server-sse

提示：確保已在 pyproject.toml 中定義了 script entry point。

配置 Claude Code

在 Claude Code 中使用 SSE server，需要在配置檔案中加入以下設定：

{
  "mcpServers": {
    "weather": {
      "url": "http://localhost:3000/sse"
    }
  }
}

與 stdio transport 不同，這裡使用 url 而非 command。

SSE vs Stdio Transport 比較

特性	Stdio Transport	SSE Transport
通訊方式	標準輸入/輸出	HTTP (SSE + POST)
啟動方式	由 client 啟動 process	獨立運行的 HTTP server
生命週期	與 client 綁定	獨立於 client
多 client 支援	否（一對一）	是（一對多）
適用場景	本地開發、簡單整合	生產環境、多用戶服務

除錯技巧

查看 Server 日誌

SSE server 的日誌會直接輸出到啟動 server 的終端機：

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:3000

測試連線

可以使用 curl 測試 SSE endpoint 是否正常運作：

curl -N http://localhost:3000/sse

注意：確保 server 已啟動且沒有其他程式佔用 port 3000。

結語

相關檔案：src/weather_server/sse_server.py:1

High Altitude Oolong

2025/10/11

MCP Weather Server：SSE Transport