C3-06 Mcp Server

from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("hr-knowledge")

@mcp.resource("employees://list")
def list_employees() -> str:
    """返回员工列表(JSON 格式)"""
    employees = [
        {"id": 1, "name": "Alice", "dept": "Sales"},
        {"id": 2, "name": "Bob", "dept": "R&D"},
    ]
    return json.dumps(employees, ensure_ascii=False)


@mcp.resource("employee://{emp_id}")
def get_employee(emp_id: int) -> str:
    """根据 ID 获取员工详情"""
    return json.dumps({"id": emp_id, "name": "Alice", "credit": 720})

@mcp.tool()
def add_employee(name: str, dept: str) -> str:
    """新增员工
    Args:
        name: 员工姓名
        dept: 部门名称
    """
    new_emp = {"id": 999, "name": name, "dept": dept}
    return json.dumps({"ok": True, "employee": new_emp}, ensure_ascii=False)


@mcp.tool()
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果: {result}"
    except Exception as e:
        return f"计算错误: {e}"

import json
from datetime import datetime
from mcp.server.fastmcp import FastMCP

# ========== 1. 创建 MCP Server ==========
mcp = FastMCP("employee-management")

# 模拟数据库
database = {
    "employees": [
        {"id": 1, "name": "张三", "dept": "技术部", "salary": 25000},
        {"id": 2, "name": "李四", "dept": "产品部", "salary": 22000},
        {"id": 3, "name": "王五", "dept": "技术部", "salary": 28000},
        {"id": 4, "name": "赵六", "dept": "销售部", "salary": 20000},
    ]
}


# ========== 2. 注册 Resources(只读) ==========
@mcp.resource("employees://list")
def list_all_employees() -> str:
    """获取所有员工列表"""
    return json.dumps(database["employees"], ensure_ascii=False, indent=2)


@mcp.resource("stats://summary")
def get_stats() -> str:
    """获取员工统计摘要"""
    total = len(database["employees"])
    avg_salary = sum(e["salary"] for e in database["employees"]) / total
    return json.dumps(
        {
            "total_employees": total,
            "average_salary": avg_salary,
            "updated_at": datetime.now().isoformat(),
        },
        ensure_ascii=False,
    )


# ========== 3. 注册 Tools(可执行) ==========
@mcp.tool()
def add_employee(name: str, dept: str, salary: int) -> str:
    """添加新员工
    Args:
        name: 员工姓名
        dept: 部门名称
        salary: 月薪
    """
    new_id = max(e["id"] for e in database["employees"]) + 1
    new_emp = {"id": new_id, "name": name, "dept": dept, "salary": salary}
    database["employees"].append(new_emp)
    return json.dumps({"success": True, "employee": new_emp}, ensure_ascii=False)


@mcp.tool()
def calculate_dept_budget(dept: str) -> str:
    """计算部门薪资预算
    Args:
        dept: 部门名称
    """
    dept_employees = [e for e in database["employees"] if e["dept"] == dept]
    total_salary = sum(e["salary"] for e in dept_employees)
    return json.dumps(
        {
            "department": dept,
            "employee_count": len(dept_employees),
            "total_monthly_salary": total_salary,
            "annual_budget": total_salary * 12,
        },
        ensure_ascii=False,
    )


# ========== 4. 启动 Server ==========
if __name__ == "__main__":
    # 默认 stdio 传输(本地)
    mcp.run(transport="stdio")

    # 生产环境用 HTTP:
    # mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

{
  "mcpServers": {
    "employee-management": {
      "command": "python",
      "args": ["/path/to/05_mcp_server.py"]
    }
  }
}

from langchain_mcp import MCPToolkit
from langgraph.prebuilt import create_react_agent

# 连接 MCP Server
toolkit = MCPToolkit.from_server("employee-management", "stdio://localhost")

# 创建 Agent(自动获得 MCP Server 暴露的所有工具)
agent = create_react_agent(model="gpt-4o", tools=toolkit.get_tools())

# 使用
result = agent.invoke({"messages": [("user", "技术部一共多少薪资预算?")]})