import json
from typing import Any, Dict, List, Optional
from mcp.types import Tool, TextContent
from tools.base import BaseTool
from utils.helpers import APIHelper, format_response
from models.schemas import Employee, EmployeeSearchParams
class EmployeeManagementTools(BaseTool):
"""员工管理工具集"""
def __init__(self):
super().__init__()
self.api_helper = APIHelper()
def get_all_tools(self) -> List[Tool]:
"""获取所有工具定义"""
return [
self.get_employee_by_id_tool(),
self.get_all_employees_tool(),
self.add_employee_tool(),
self.update_employee_tool(),
self.delete_employee_tool(),
self.search_employees_tool(),
self.get_employees_by_level_tool(),
self.get_active_employees_tool(),
self.sync_employees_tool()
]
def get_employee_by_id_tool(self) -> Tool:
"""根据ID查询员工工具"""
return Tool(
name="get_employee_by_id",
description="根据员工ID获取单个员工的详细信息",
inputSchema={
"type": "object",
"properties": {
"userId": {
"type": "integer",
"description": "员工ID"
}
},
"required": ["userId"]
}
)
def get_all_employees_tool(self) -> Tool:
"""查询所有员工工具"""
return Tool(
name="get_all_employees",
description="获取系统中所有员工的列表信息",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
)
def add_employee_tool(self) -> Tool:
"""新增员工工具"""
return Tool(
name="add_employee",
description="添加新的员工信息到系统中",
inputSchema={
"type": "object",
"properties": {
"firstName": {
"type": "string",
"description": "员工名字"
},
"lastName": {
"type": "string",
"description": "员工姓氏"
},
"salary": {
"type": "number",
"description": "员工工资"
},
"currency": {
"type": "string",
"description": "货币单位",
"enum": ["CNY", "USD", "EUR"]
},
"birthdate": {
"type": "string",
"description": "员工生日 (YYYY-MM-DD格式)"
},
"isActive": {
"type": "boolean",
"description": "在职状态"
},
"level": {
"type": "string",
"description": "员工级别"
}
},
"required": ["firstName", "lastName", "salary", "birthdate", "level"]
}
)
def update_employee_tool(self) -> Tool:
"""更新员工信息工具"""
return Tool(
name="update_employee",
description="根据员工ID更新员工信息",
inputSchema={
"type": "object",
"properties": {
"userId": {
"type": "integer",
"description": "员工ID"
},
"firstName": {
"type": "string",
"description": "员工名字"
},
"lastName": {
"type": "string",
"description": "员工姓氏"
},
"salary": {
"type": "number",
"description": "员工工资"
},
"currency": {
"type": "string",
"description": "货币单位",
"enum": ["CNY", "USD", "EUR"]
},
"birthdate": {
"type": "string",
"description": "员工生日 (YYYY-MM-DD格式)"
},
"isActive": {
"type": "boolean",
"description": "在职状态"
},
"level": {
"type": "string",
"description": "员工级别"
}
},
"required": ["userId"]
}
)
def delete_employee_tool(self) -> Tool:
"""删除员工工具"""
return Tool(
name="delete_employee",
description="根据员工ID删除员工信息",
inputSchema={
"type": "object",
"properties": {
"userId": {
"type": "integer",
"description": "员工ID"
}
},
"required": ["userId"]
}
)
def search_employees_tool(self) -> Tool:
"""条件搜索员工工具"""
return Tool(
name="search_employees",
description="根据姓氏、在职状态、级别等条件搜索员工",
inputSchema={
"type": "object",
"properties": {
"lastName": {
"type": "string",
"description": "员工姓氏"
},
"isActive": {
"type": "boolean",
"description": "在职状态"
},
"level": {
"type": "string",
"description": "员工级别"
}
},
"required": []
}
)
def get_employees_by_level_tool(self) -> Tool:
"""根据级别查询员工工具"""
return Tool(
name="get_employees_by_level",
description="根据员工级别查询所有符合条件的员工",
inputSchema={
"type": "object",
"properties": {
"level": {
"type": "string",
"description": "员工级别"
}
},
"required": ["level"]
}
)
def get_active_employees_tool(self) -> Tool:
"""查询在职员工工具"""
return Tool(
name="get_active_employees",
description="获取所有在职状态的员工列表",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
)
def sync_employees_tool(self) -> Tool:
"""同步员工数据工具"""
return Tool(
name="sync_employees",
description="从JSON文件同步员工数据到数据库",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
)
async def execute(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""执行工具"""
try:
if tool_name == "get_employee_by_id":
return await self._get_employee_by_id(arguments)
elif tool_name == "get_all_employees":
return await self._get_all_employees()
elif tool_name == "add_employee":
return await self._add_employee(arguments)
elif tool_name == "update_employee":
return await self._update_employee(arguments)
elif tool_name == "delete_employee":
return await self._delete_employee(arguments)
elif tool_name == "search_employees":
return await self._search_employees(arguments)
elif tool_name == "get_employees_by_level":
return await self._get_employees_by_level(arguments)
elif tool_name == "get_active_employees":
return await self._get_active_employees()
elif tool_name == "sync_employees":
return await self._sync_employees()
else:
return self.format_error_response(f"未知工具: {tool_name}")
except Exception as e:
return self.format_error_response(str(e))
async def _get_employee_by_id(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""根据ID查询员工"""
userId = arguments.get("userId")
if not userId:
return self.format_error_response("员工ID不能为空")
response = await self.api_helper.make_request("GET", f"/api/employees/{userId}")
return self.format_success_response(response, "查询员工成功")
async def _get_all_employees(self) -> Dict[str, Any]:
"""查询所有员工"""
response = await self.api_helper.make_request("GET", "/api/employees")
return self.format_success_response(response, "查询所有员工成功")
async def _add_employee(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""新增员工"""
response = await self.api_helper.make_request("POST", "/api/employees", data=arguments)
return self.format_success_response(response, "新增员工成功")
async def _update_employee(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""更新员工信息"""
userId = arguments.pop("userId")
response = await self.api_helper.make_request("PUT", f"/api/employees/{userId}", data=arguments)
return self.format_success_response(response, "更新员工成功")
async def _delete_employee(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""删除员工"""
userId = arguments.get("userId")
if not userId:
return self.format_error_response("员工ID不能为空")
response = await self.api_helper.make_request("DELETE", f"/api/employees/{userId}")
return self.format_success_response(response, "删除员工成功")
async def _search_employees(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""条件搜索员工"""
# 过滤空值参数
params = {k: v for k, v in arguments.items() if v is not None}
response = await self.api_helper.make_request("GET", "/api/employees/search", params=params)
return self.format_success_response(response, "搜索员工成功")
async def _get_employees_by_level(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""根据级别查询员工"""
level = arguments.get("level")
if not level:
return self.format_error_response("员工级别不能为空")
response = await self.api_helper.make_request("GET", f"/api/employees/level/{level}")
return self.format_success_response(response, "根据级别查询员工成功")
async def _get_active_employees(self) -> Dict[str, Any]:
"""查询在职员工"""
response = await self.api_helper.make_request("GET", "/api/employees/active")
return self.format_success_response(response, "查询在职员工成功")
async def _sync_employees(self) -> Dict[str, Any]:
"""同步员工数据"""
response = await self.api_helper.make_request("GET", "/api/employees/sync")
return self.format_success_response(response, "同步员工数据成功")
