内容
# MCP Server with Server-Sent Events (SSE)
该项目实现了一个支持服务器推送事件（Server-Sent Events, SSE）的模型上下文协议（Model Context Protocol, MCP）服务器，允许服务器与客户端之间进行实时通信。
## 特性
- 支持工具和资源的 MCP 服务器
- 服务器推送事件（SSE）用于实时更新
- 跨域请求的 CORS 中间件
- 内置工具：
- `add(a: int, b: int)`: 添加两个数字
- `get_metrics(question: str)`: 返回给定问题的指标
- `greeting://{name}`: 返回个性化问候
## 安装
1. 创建并激活一个 Python 虚拟环境：
```bash
python -m venv venv
source venv/bin/activate # 在 Unix/macOS 上
# 或者
.\venv\Scripts\activate # 在 Windows 上
```
2. 安装依赖：
```bash
uv pip install -r requirements.txt
```
## 运行服务器
1. 启动服务器：
```bash
uvicorn server_sse:starlette_app --host 0.0.0.0 --port 8000 --reload
```
服务器将在以下地址可用：
- HTTP 服务器: http://localhost:8000
- SSE 端点: /mcp/sse 和 /sse
- 消息端点: /mcp/messages 和 /messages
## 使用客户端
### Python 客户端示例
以下是如何在 Python 中使用 SSE 客户端的示例：
```python
import aiohttp
import asyncio
import json
import logging
class MCPClient:
def __init__(self, base_url="http://localhost:8000"):
self.base_url = base_url
self.messages_url = f"{base_url}/mcp/messages"
self.sse_url = f"{base_url}/mcp/sse"
async def list_tools(self):
"""列出 MCP 服务器上所有可用的工具。"""
try:
introspection_message = {
"type": "request",
"action": "introspect",
"data": {}
}
async with aiohttp.ClientSession() as session:
async with session.post(self.messages_url, json=introspection_message) as response:
if response.status == 200:
data = await response.json()
return data.get("data", {})
else:
error_text = await response.text()
logging.error(f"获取工具失败。状态码: {response.status}")
logging.error(f"错误响应: {error_text}")
return None
except Exception as e:
logging.error(f"列出工具时出错: {e}")
return None
async def add_numbers(self, a: int, b: int):
"""使用 add 工具求和两个数字。"""
message = {
"type": "request",
"action": "execute",
"tool": "add",
"data": {"a": a, "b": b}
}
async with aiohttp.ClientSession() as session:
async with session.post(self.messages_url, json=message) as response:
result = await response.json()
return result.get("data")
async def get_metrics(self, question: str):
"""获取特定问题的指标。"""
message = {
"type": "request",
"action": "execute",
"tool": "get_metrics",
"data": {"question": question}
}
async with aiohttp.ClientSession() as session:
async with session.post(self.messages_url, json=message) as response:
result = await response.json()
return result.get("data")
async def get_greeting(self, name: str):
"""获取个性化问候。"""
message = {
"type": "request",
"action": "get",
"resource": f"greeting://{name}",
"data": {}
}
async with aiohttp.ClientSession() as session:
async with session.post(self.messages_url, json=message) as response:
result = await response.json()
return result.get("data")
```
### 使用客户端
```python
async def main():
client = MCPClient()
# 列出可用工具
tools = await client.list_tools()
print("\n可用工具和资源:")
print(json.dumps(tools, indent=2))
# 添加两个数字
result = await client.add_numbers(5, 3)
print("\n5 + 3 =", result) # 输出: 8
# 获取指标
metrics = await client.get_metrics("特性 X 的表现如何？")
print("\n指标:", json.dumps(metrics, indent=2))
# 获取问候
greeting = await client.get_greeting("Alice")
print("\n问候:", greeting) # 输出: Hello, Alice!
if __name__ == "__main__":
asyncio.run(main())
```
### JavaScript/浏览器客户端示例
```javascript
// 连接到 SSE 流
const eventSource = new EventSource('http://localhost:8000/mcp/sse');
// 监听消息
eventSource.onmessage = function(event) {
const data = JSON.parse(event.data);
console.log('接收到:', data);
};
// 发送消息以执行 add 工具
async function addNumbers(a, b) {
const response = await fetch('http://localhost:8000/mcp/messages', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
type: 'request',
action: 'execute',
tool: 'add',
data: { a, b }
})
});
const result = await response.json();
return result.data; // 仅返回结果数字
}
// 获取问候
async function getGreeting(name) {
const response = await fetch('http://localhost:8000/mcp/messages', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
type: 'request',
action: 'get',
resource: `greeting://${name}`,
data: {}
})
});
const result = await response.json();
return result.data; // 仅返回问候字符串
}
// 在 JavaScript addNumbers 函数后添加
async function getMetrics(question) {
const response = await fetch('http://localhost:8000/mcp/messages', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({
type: 'request',
action: 'execute',
tool: 'get_metrics',
data: { question }
})
});
const result = await response.json();
return result.data; // 返回指标对象
}
// 更新示例用法：
async function example() {
// 添加数字
const sum = await addNumbers(5, 3);
console.log('5 + 3 =', sum); // 输出: 8
// 获取指标
const metrics = await getMetrics('特性 X 的表现如何？');
console.log('指标:', metrics);
// 获取问候
const greeting = await getGreeting('Alice');
console.log('问候:', greeting); // 输出: Hello, Alice!
}
```
## API 文档
### 工具
#### 添加工具
- 名称: `add`
- 描述: 添加两个数字
- 参数:
- `a`: 整数
- `b`: 整数
- 返回: 整数（a 和 b 的和）
#### 获取指标工具
- 名称: `get_metrics`
- 描述: 获取与特定问题相关的指标
- 参数:
- `question`: 字符串
- 返回: 具有以下结构的对象：
```json
{
"metrics": {
"question": "string",
"total_responses": "number",
"average_score": "number",
"completion_rate": "string",
"time_spent": "string",
"difficulty_level": "string",
"success_rate": "string",
"feedback_count": "number"
},
"timestamp": "string"
}
```
### 资源
#### 问候资源
- 模式: `greeting://{name}`
- 描述: 获取个性化问候
- 参数:
- `name`: 字符串
- 返回: 字符串（个性化问候消息）
### 消息格式
#### 请求格式
```json
{
"type": "request",
"action": "execute|get|introspect",
"tool": "tool_name", // 对于 execute
"resource": "resource_pattern", // 对于 get
"data": {
// 参数
}
}
```
#### 响应格式
```json
{
"type": "response",
"data": "result" // 实际结果值，不再另一个对象中包装
}
```
## 错误处理
服务器返回状态码为 500 的错误响应和错误消息：
```json
{
"error": "错误消息描述"
}
```
## 日志记录
服务器日志同时写入控制台和 `logs/server_sse.log`。日志级别设置为 DEBUG，以获取详细信息。
连接信息
猜你喜欢
暂无推荐服务器