使用说明
MCP Hello Server（Python）
使用FastAPI实现的最小MCP风格服务器。它提供：
GET /health-简单健康检查POST /mcp-接受一个较小的JSON有效负载（模型+消息），并返回一个简单的助手响应
此项目有意设计得较小，因此您可以在本地运行它并对其进行扩展以实现真正的模型上下文协议服务器。
快速入门（PowerShell）：
Set-Location 'C:\Users\Cyphanet\mcp_hello_server_py'
python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
pip install -r requirements.txt
uvicorn app:app --host 127.0.0.1 --port 8000
使用以下工具测试MCP端点 curl 或 http 客户：
curl -X POST "http://127.0.0.1:8000/mcp" -H "Content-Type: application/json" -d '{"model":"mcp-hello-1","messages":[{"role":"user","content":"Hello"}]}'
运行测试：
pytest -q
扩展MCP端点
这 /mcp 端点当前回显用户消息。为了实现真正的MCP行为，以下是常见的模式：
1.流媒体响应
使用FastAPI StreamingResponse 在块到达时发送块：
from fastapi.responses import StreamingResponse
import asyncio
async def generate_streaming_response(messages):
"""Stream tokens as they are generated."""
for token in ["Hello", " ", "from", " ", "MCP", "!"]:
yield f"data: {token}\n"
await asyncio.sleep(0.1)
@app.post("/mcp/stream")
async def mcp_stream(req: MCPRequest):
return StreamingResponse(
generate_streaming_response(req.messages),
media_type="text/event-stream"
)
2.模型挂钩
创建一个钩子系统，在生成之前/之后注入自定义逻辑：
from typing import Callable
hooks: dict[str, list[Callable]] = {"pre_generate": [], "post_generate": []}
def register_hook(event: str, fn: Callable):
"""Register a hook function."""
hooks[event].append(fn)
async def process_mcp_request(req: MCPRequest):
# Pre-generation hooks
for hook in hooks["pre_generate"]:
await hook(req)
# Generate response (your model call here)
response = f"Echo: {req.messages}"
# Post-generation hooks
for hook in hooks["post_generate"]:
await hook(response)
return response
@app.post("/mcp")
async def mcp_endpoint(req: MCPRequest):
content = await process_mcp_request(req)
return {
"model": req.model,
"choices": [{"message": {"role": "assistant", "content": content}}]
}
3.整合外部模型
将echo逻辑替换为对OpenAI、Claude或您自己的模型的调用：
from openai import AsyncOpenAI
client = AsyncOpenAI(api_key="sk-...")
@app.post("/mcp")
async def mcp_endpoint(req: MCPRequest):
response = await client.chat.completions.create(
model="gpt-4",
messages=[{"role": m.role, "content": m.content} for m in req.messages]
)
return {
"model": req.model,
"choices": [{"message": {"role": "assistant", "content": response.choices[0].message.content}}]
}
4.请求验证和错误处理
扩展Pydantic模型并添加异常处理程序：
from fastapi import HTTPException
from pydantic import validator
class MCPRequest(BaseModel):
model: str
messages: List[Message]
@validator("messages")
def messages_not_empty(cls, v):
if not v:
raise ValueError("messages cannot be empty")
return v
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
return JSONResponse(
status_code=400,
content={"detail": str(exc)}
)
从echo逻辑开始，一次添加一个模式，并使用提供的pytest套件进行测试。
