🚀 基於終端的聊天客戶端與MCP服務器集成
本項目展示瞭如何構建一個基於終端的聊天客戶端界面。該客戶端可連接到MCP服務器,並集成了OpenAI API。此外,項目還包含一個簡單的天氣服務,作為MCP功能的示例。
🚀 快速開始
先決條件
- Python 3.8 或更高版本
- UV 包管理器(一種快速、可靠的 Python 包安裝和解析工具)
安裝
1. 安裝 UV
UV 是一個現代的 Python 包管理器,相較於傳統工具(如 pip),它能顯著提升性能。它使用 Rust 編寫,具備以下特性:
- 更快的包安裝速度
- 可靠的依賴項解析
- 內置虛擬環境管理
- 與現有 Python 工具兼容
要安裝 UV,請運行以下命令:
curl -LsSf https://astral.sh/uv/install.sh | sh
2. 項目設置
- 初始化新項目:
uv init
- 創建並激活虛擬環境:
uv venv
source .venv/bin/activate
.venv\Scripts\activate
- 安裝所需的包:
uv pip install httpx mcp[cli] openai python-dotenv
使用說明
- 啟動天氣服務器:
python weather.py
- 啟動聊天客戶端並連接到天氣服務器:
python client.py weather.py
- 在聊天界面中輸入消息,例如:
- “What's the weather in California?”(加利福尼亞的天氣如何?)
- “Are there any alerts in New York?”(紐約有警報嗎?)
✨ 主要特性
- 即時聊天界面與 OpenAI 集成
- MCP 服務器集成以實現可擴展功能
- 天氣服務帶有警報和預報
- 異步操作以提高性能
- 正確的錯誤處理和資源清理
- 環境變量配置用於 API 密鑰
📦 安裝指南
安裝 UV
UV 是一個現代的 Python 包管理器,提供比傳統工具(如 pip)顯著的性能改進。它用 Rust 編寫,並提供:
- 更快的包安裝
- 可靠的依賴項解析
- 內置虛擬環境管理
- 與現有 Python 工具兼容
要安裝 UV,請運行:
curl -LsSf https://astral.sh/uv/install.sh | sh
項目設置
- 初始化新項目:
uv init
- 創建並激活虛擬環境:
uv venv
source .venv/bin/activate
.venv\Scripts\activate
- 安裝所需的包:
uv pip install httpx mcp[cli] openai python-dotenv
💻 使用示例
基礎用法
啟動天氣服務器
python weather.py
啟動聊天客戶端並連接到天氣服務器
python client.py weather.py
高級用法
在聊天界面中輸入消息,例如:
# 查詢加利福尼亞的天氣
What's the weather in California?
# 查詢紐約是否有警報
Are there any alerts in New York?
📚 詳細文檔
項目結構和實現指南
該項目由兩個主要組件組成:一個聊天客戶端(client.py
)和一個天氣服務(weather.py
)。以下為您詳細介紹每個組件的構建方式和功能。
構建聊天客戶端(client.py)
聊天客戶端是一個異步 Python 應用程序,連接到 MCP 服務器並集成 OpenAI API。以下是它的構建步驟:
- 導入和設置
import asyncio
import os
import sys
from typing import Optional
from contextlib import AsyncExitStack
from dotenv import load_dotenv
import openai
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
- `asyncio`:用於異步編程
- `os` 和 `sys`:用於與操作系統交互
- `typing.Optional`:用於類型提示
- `AsyncExitStack`:用於管理異步上下文
- `dotenv`:用於加載環境變量
- `openai`:OpenAI API 客戶端
- `mcp`:MCP 協議庫
- 聊天客戶端類
class ChatClient:
def __init__(self, server_address):
self.server_address = server_address
self.loop = None
self.client_session = None
async def run(self):
self.loop = asyncio.get_event_loop()
self.client_session = await ClientSession.create(
loop=self.loop,
server_address=self.server_address
)
await self.client_session.connect()
async def send_message(self, message):
if self.client_session.is_connected:
await self.client_session.send(message)
async def close(self):
if self.client_session:
await self.client_session.close()
- 使用聊天客戶端
import logging
logging.basicConfig(level=logging.INFO)
client = ChatClient('weather.py')
try:
await client.run()
while True:
message = input('> ')
await client.send_message(message)
except KeyboardInterrupt:
await client.close()
構建天氣服務(weather.py)
天氣服務是一個 MCP 服務器,提供天氣相關功能:
- 導入和設置
import asyncio
from mcp_server import MCPServer
from weather_api import WeatherAPI
- 天氣服務類
class WeatherService:
def __init__(self):
self.server = None
self.weather_api = WeatherAPI()
async def start(self, host='127.0.0.1', port=8000):
self.server = await MCPServer.create(
host=host,
port=port,
handlers={
'get_weather': self.get_weather,
'get_alerts': self.get_alerts
}
)
await self.server.start()
async def get_weather(self, location):
return await self.weather_api.get_weather(location)
async def get_alerts(self, location):
return await self.weather_api.get_alerts(location)
- 運行天氣服務
import logging
logging.basicConfig(level=logging.INFO)
service = WeatherService()
try:
await service.start()
except KeyboardInterrupt:
await service.stop()
項目結構
chat_client/
├── client.py # 聊天客戶端實現
├── requirements.txt # 依賴項列表
└── README.md # 項目說明
weather_service/
├── weather.py # 天氣服務實現
├── weather_api.py # 第三方天氣 API 接口
└── setup.py # 安裝腳本
🔧 技術細節
該項目使用 Python 3.8 或更高版本進行開發,藉助 UV 包管理器來管理項目依賴。項目採用異步編程模型,使用 asyncio
庫實現異步操作,提高了程序的性能。聊天客戶端通過 mcp
庫連接到 MCP 服務器,並集成了 OpenAI API 實現智能交互。天氣服務則通過 mcp_server
庫搭建 MCP 服務器,提供天氣相關功能。同時,項目使用 python-dotenv
庫來管理環境變量,確保 API 密鑰等敏感信息的安全。
📄 許可證
文檔中未提及相關許可證信息。
貢獻
歡迎提交問題和增強請求!