#!/usr/bin/env python3
"""
MCP Hello World 服务器 - HTTP 版本
支持远程部署，可通过HTTP/HTTPS访问
兼容Manus平台注册要求
"""
import asyncio
import json
import os
from typing import Any, Sequence, Dict
from mcp.server import Server
from mcp.types import (
Tool,
TextContent,
ImageContent,
EmbeddedResource,
JSONRPCRequest,
JSONRPCResponse,
)
from pydantic import BaseModel, Field, field_validator
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse, RedirectResponse
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 定义输入参数模型，包含约束
class HelloWorldInput(BaseModel):
"""Hello World工具的输入参数"""
name: str = Field(
default="World",
description="要问候的名字",
min_length=1,
max_length=100,
pattern="^[a-zA-Z0-9\\s\\u4e00-\\u9fa5]+$"
)
language: str = Field(
default="zh",
description="问候语言：zh(中文) 或 en(英文)",
pattern="^(zh|en)$"
)
count: int = Field(
default=1,
description="重复次数",
ge=1,
le=10
)
@field_validator('name')
@classmethod
def validate_name(cls, v: str) -> str:
"""验证并清理名字"""
v = v.strip()
if not v:
raise ValueError("名字不能为空")
if len(v) > 100:
raise ValueError("名字长度不能超过100个字符")
return v
@field_validator('language')
@classmethod
def validate_language(cls, v: str) -> str:
"""验证语言代码"""
if v not in ['zh', 'en']:
raise ValueError("语言必须是 'zh' 或 'en'")
return v
# 创建MCP服务器实例
app = Server("hello-world-mcp-http")
# 定义工具列表（可复用）
async def get_tools_list() -> list[Tool]:
"""
获取所有可用的工具列表
"""
return [
Tool(
name="hello_world",
description="一个简单的Hello World工具，支持多语言问候",
inputSchema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "要问候的名字（1-100个字符，支持中文）",
"default": "World",
"minLength": 1,
"maxLength": 100
},
"language": {
"type": "string",
"description": "问候语言：zh(中文) 或 en(英文)",
"enum": ["zh", "en"],
"default": "zh"
},
"count": {
"type": "integer",
"description": "重复次数（1-10次）",
"minimum": 1,
"maximum": 10,
"default": 1
}
},
"required": []
}
)
]
@app.list_tools()
async def list_tools() -> list[Tool]:
"""
列出所有可用的工具（MCP协议处理器）
"""
return await get_tools_list()
async def call_tool_handler(name: str, arguments: dict[str, Any] | None) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""
工具调用处理函数（可直接调用）
"""
if name != "hello_world":
raise ValueError(f"未知的工具: {name}")
try:
if arguments is None:
arguments = {}
input_data = HelloWorldInput(**arguments)
greetings = {
"zh": f"你好，{input_data.name}！",
"en": f"Hello, {input_data.name}!"
}
greeting = greetings.get(input_data.language, greetings["zh"])
result_lines = []
for i in range(input_data.count):
result_lines.append(f"{i+1}. {greeting}")
result = "\n".join(result_lines)
metadata = {
"language": input_data.language,
"name": input_data.name,
"count": input_data.count
}
return [
TextContent(
type="text",
text=f"{result}\n\n[元数据: {json.dumps(metadata, ensure_ascii=False)}]"
)
]
except ValueError as e:
error_msg = f"参数验证失败: {str(e)}"
return [
TextContent(
type="text",
text=f"错误: {error_msg}\n\n提示: 请检查输入参数是否符合要求。"
)
]
except Exception as e:
error_msg = f"处理请求时发生错误: {str(e)}"
return [
TextContent(
type="text",
text=f"错误: {error_msg}"
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any] | None) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
"""MCP协议工具调用（用于stdio模式）"""
return await call_tool_handler(name, arguments)
# 创建FastAPI应用用于HTTP服务
fastapi_app = FastAPI(
title="Hello World MCP Server",
version="1.0.0",
description="MCP Hello World 服务器 - HTTP版本，支持远程部署和Manus注册"
)
# 添加CORS支持（生产环境应限制特定域名）
fastapi_app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 生产环境应设置具体域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 存储活跃的MCP会话
active_sessions: Dict[str, Any] = {}
@fastapi_app.get("/")
async def root():
"""根路径重定向到 /mcp"""
return RedirectResponse(url="/mcp")
@fastapi_app.get("/mcp")
async def mcp_info():
"""MCP服务器信息端点（GET）"""
return {
"name": "hello-world-mcp-http",
"version": "1.0.0",
"status": "running",
"description": "MCP Hello World 服务器 - HTTP版本",
"protocol": "MCP",
"endpoints": {
"health": "/health",
"mcp": "/mcp",
"tools": "/tools"
}
}
@fastapi_app.get("/health")
async def health():
"""健康检查"""
return {
"status": "healthy",
"service": "hello-world-mcp"
}
@fastapi_app.get("/tools")
async def list_tools_endpoint():
"""列出所有可用工具的HTTP端点"""
tools = await get_tools_list()
return {
"tools": [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.inputSchema
}
for tool in tools
]
}
@fastapi_app.post("/mcp")
async def mcp_endpoint(request: Request):
"""
MCP协议端点，处理JSON-RPC请求
支持Manus平台的MCP协议通信
"""
try:
body = await request.json()
logger.info(f"收到MCP请求: {body}")
# 处理MCP JSON-RPC请求
if "method" not in body:
raise HTTPException(status_code=400, detail="缺少method字段")
method = body["method"]
params = body.get("params", {})
request_id = body.get("id")
# 处理不同的MCP方法
if method == "tools/list":
tools = await get_tools_list()
return JSONResponse({
"jsonrpc": "2.0",
"id": request_id,
"result": {
"tools": [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.inputSchema
}
for tool in tools
]
}
})
elif method == "tools/call":
tool_name = params.get("name")
arguments = params.get("arguments", {})
if not tool_name:
raise HTTPException(status_code=400, detail="缺少工具名称")
# 直接调用工具处理函数
result = await call_tool_handler(tool_name, arguments)
# 格式化返回结果
return JSONResponse({
"jsonrpc": "2.0",
"id": request_id,
"result": {
"content": [
{
"type": content.type,
"text": content.text if hasattr(content, "text") else str(content)
}
for content in result
]
}
})
elif method == "initialize":
# 初始化MCP会话
return JSONResponse({
"jsonrpc": "2.0",
"id": request_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"serverInfo": {
"name": "hello-world-mcp",
"version": "1.0.0"
}
}
})
else:
raise HTTPException(status_code=400, detail=f"未知的方法: {method}")
except HTTPException:
raise
except Exception as e:
logger.error(f"处理MCP请求时出错: {e}", exc_info=True)
return JSONResponse(
status_code=500,
content={
"jsonrpc": "2.0",
"id": body.get("id") if "body" in locals() else None,
"error": {
"code": -32603,
"message": f"内部错误: {str(e)}"
}
}
)
def main():
"""
启动HTTP服务器
"""
host = os.getenv("HOST", "0.0.0.0")
port = int(os.getenv("PORT", "8000"))
print(f"启动 MCP Hello World 服务器...")
print(f"服务器地址: http://{host}:{port}")
print(f"健康检查: http://{host}:{port}/health")
print(f"SSE端点: http://{host}:{port}/sse")
print(f"按 Ctrl+C 停止服务器")
uvicorn.run(
fastapi_app,
host=host,
port=port,
log_level="info"
)
if __name__ == "__main__":
main()
