#!/usr/bin/env python3
"""
MCP Hello World 服务器
实现一个简单的hello world工具，展示MCP协议的基本用法
"""
import asyncio
import json
import sys
from typing import Any, Sequence
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
Tool,
TextContent,
ImageContent,
EmbeddedResource,
LoggingLevel
)
from pydantic import BaseModel, Field, field_validator
# 定义输入参数模型，包含约束
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 # 最多重复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")
@app.list_tools()
async def list_tools() -> 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.call_tool()
async def call_tool(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 = {}
# 使用Pydantic模型进行验证
input_data = HelloWorldInput(**arguments)
# 生成问候语
greetings = {
"zh": f"你好，{input_data.name}！",
"en": f"Hello, {input_data.name}!"
}
greeting = greetings.get(input_data.language, greetings["zh"])
# 根据count重复问候
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}"
)
]
async def main():
"""
主函数：启动MCP服务器
"""
# 使用stdio传输方式运行服务器
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
