ai-agent/06_mcp/my_mcp_server.py

# my_mcp_server.py —— 一个极简的自定义 MCP 服务
from mcp.server.fastmcp import FastMCP
from datetime import datetime
from urllib.parse import unquote

# ============================================================
# 第一步：创建 FastMCP 实例，相当于初始化一个 Server 框架
# ============================================================
mcp = FastMCP(name="my-tools")


# ============================================================
# 第二步：用 @mcp.tool() 装饰器注册工具
# 装饰器会自动提取函数的 type hints 和 docstring 作为工具描述
# ============================================================

@mcp.tool()
def get_current_time() -> str:
    """
    获取当前日期和时间，返回格式友好的字符串。

    适用场景：当用户问"现在几点"或"今天几号"时调用。
    """
    now = datetime.now()
    return f"现在是 {now.year}年{now.month}月{now.day}日，{now.strftime('%H:%M:%S')}"


@mcp.tool()
def calculate_bmi(weight_kg: float, height_cm: float) -> str:
    """
    根据体重（公斤）和身高（厘米）计算 BMI 指数。

    参数：
    - weight_kg: 体重，单位公斤
    - height_cm: 身高，单位厘米

    返回：BMI 数值 + 健康评级
    """
    height_m = height_cm / 100
    bmi = weight_kg / (height_m ** 2)

    # 按中国标准分级
    if bmi < 18.5:
        level = "偏瘦，多吃点"
    elif bmi < 24:
        level = "正常，继续保持"
    elif bmi < 28:
        level = "偏胖，注意饮食"
    else:
        level = "该运动了兄弟"

    return f"BMI = {bmi:.1f}，{level}"


# ============================================================
# 第三步：用 @mcp.resource() 注册静态资源
# 资源是"模板化"的数据端点，适合暴露文章、配置、文件内容等
# URI 中用 {变量名} 定义路径参数
# ============================================================

# 模拟一个简易知识库
KNOWLEDGE_BASE = {
    "请假流程": "OA 系统提交 → 直属领导审批 → HR 备案。年假需提前 3 天申请。",
    "报销标准": "市内交通实报实销（上限 100/天），住宿标准 350/晚（一线城市 450/晚）。",
    "入职材料": "身份证原件 + 复印件、学历学位证、体检报告、离职证明、银行卡。",
}

@mcp.resource("knowledge://{topic}")
def query_knowledge(topic: str) -> str:
    """
    查询公司内部知识库，返回指定主题的说明文档。

    调用方式（Client 端）：session.read_resource("knowledge://报销标准")
    """
    # MCP 协议中 URI 的非 ASCII 字符会被 Pydantic AnyUrl 自动
    # URL 编码（如 "请假流程" → "%E8%AF%B7%E5%81%87%E6%B5%81%E7%A8%8B"），
    # 但 FastMCP 的 ResourceTemplate.matches() 提取参数时不会自动解码，
    # 所以这里需要手动 unquote。
    topic = unquote(topic)
    print(f"查询知识库：{topic}")
    result = KNOWLEDGE_BASE.get(topic, f"没找到关于「{topic}」的记录，请联系 HR 更新知识库。")
    return result


# ============================================================
# 第四步：启动 Server
# ============================================================
if __name__ == "__main__":
    # SSE 模式：作为独立 HTTP 服务运行，适合远程调用和调试
    mcp.settings.host = "0.0.0.0"
    mcp.settings.port = 8000
    mcp.run(transport="sse")