以下是如何使用 Python 執行經典 HTTP 串流伺服器與客戶端,以及 MCP 串流伺服器與客戶端。
- 您將設置一個 MCP 伺服器,該伺服器在處理項目時向客戶端串流進度通知。
- 客戶端將即時顯示每個通知。
- 本指南涵蓋了前置條件、設置、執行以及故障排除。
- Python 3.9 或更新版本
mcpPython 套件(使用pip install mcp安裝)
-
克隆存儲庫或下載解決方案文件。
git clone https://github.com/microsoft/mcp-for-beginners
-
建立並啟用虛擬環境(建議):
python -m venv venv .\venv\Scripts\Activate.ps1 # On Windows # or source venv/bin/activate # On Linux/macOS
-
安裝所需的依賴項:
pip install "mcp[cli]" fastapi requests
-
移動到解決方案目錄:
cd 03-GettingStarted/06-http-streaming/solution
-
啟動經典 HTTP 串流伺服器:
python server.py
-
伺服器將啟動並顯示:
Starting FastAPI server for classic HTTP streaming... INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
-
打開一個新的終端(啟用相同的虛擬環境並進入目錄):
cd 03-GettingStarted/06-http-streaming/solution python client.py
-
您應該會看到串流消息依序打印:
Running classic HTTP streaming client... Connecting to http://localhost:8000/stream with message: hello --- Streaming Progress --- Processing file 1/3... Processing file 2/3... Processing file 3/3... Here's the file content: hello --- Stream Ended ---
- 移動到解決方案目錄:
cd 03-GettingStarted/06-http-streaming/solution
- 使用 streamable-http 傳輸啟動 MCP 伺服器:
python server.py mcp
- 伺服器將啟動並顯示:
Starting MCP server with streamable-http transport... INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
- 打開一個新的終端(啟用相同的虛擬環境並進入目錄):
cd 03-GettingStarted/06-http-streaming/solution python client.py mcp
- 您應該會看到通知在伺服器處理每個項目時即時打印:
Running MCP client... Starting client... Session ID before init: None Session ID after init: a30ab7fca9c84f5fa8f5c54fe56c9612 Session initialized, ready to call tools. Received message: root=LoggingMessageNotification(...) NOTIFICATION: root=LoggingMessageNotification(...) ... Tool result: meta=None content=[TextContent(type='text', text='Processed files: file_1.txt, file_2.txt, file_3.txt | Message: hello from client')]
- 使用 FastMCP 建立 MCP 伺服器。
- 定義一個工具來處理列表,並使用
ctx.info()或ctx.log()發送通知。 - 使用
transport="streamable-http"啟動伺服器。 - 實現一個客戶端,使用消息處理器即時顯示通知。
- 伺服器使用異步函數和 MCP 上下文來發送進度更新。
- 客戶端實現了一個異步消息處理器來打印通知和最終結果。
- 使用
async/await進行非阻塞操作。 - 在伺服器和客戶端中始終處理異常以提高穩定性。
- 使用多個客戶端進行測試以觀察即時更新。
- 如果遇到錯誤,請檢查您的 Python 版本並確保所有依賴項已安裝。
免責聲明:
本文件使用 AI 翻譯服務 Co-op Translator 進行翻譯。雖然我們致力於提供準確的翻譯,但請注意,自動翻譯可能包含錯誤或不準確之處。原始文件的母語版本應被視為權威來源。對於關鍵信息,建議使用專業人工翻譯。我們對因使用此翻譯而引起的任何誤解或錯誤解釋不承擔責任。