| """MCP Tool definitions for HR Onboarding/Offboarding environment. |
| |
| Each tool is a callable that takes parameters and operates on the WorldState. |
| Tools are designed to mirror realistic enterprise HR/IT APIs. |
| """ |
|
|
| import json |
| from typing import Any, Callable |
| try: |
| from .world import WorldState |
| except ImportError: |
| from world import WorldState |
|
|
|
|
| TOOL_DEFINITIONS = [ |
| { |
| "name": "hr_create_employee", |
| "description": "Create a new employee record in the HR system. Requires name, department, level, and role. Returns the created employee record with generated emp_id.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "name": {"type": "string", "description": "Full name of the employee"}, |
| "department": {"type": "string", "description": "Department name (Engineering, Product, Marketing, Sales, Finance, HR, Data Science, Security)"}, |
| "level": {"type": "string", "description": "Employee level (L1-L6)"}, |
| "role": {"type": "string", "description": "Job title/role"}, |
| "manager_id": {"type": "string", "description": "Employee ID of the manager"}, |
| "is_contractor": {"type": "boolean", "description": "Whether this is a contractor (default: false)"}, |
| "location": {"type": "string", "description": "Office location"}, |
| "date_of_joining": {"type": "string", "description": "Start date (YYYY-MM-DD)"}, |
| }, |
| "required": ["name", "department", "level", "role"], |
| }, |
| }, |
| { |
| "name": "hr_read_employee", |
| "description": "Look up an employee by their employee ID or email address. Returns full employee record.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "emp_id": {"type": "string", "description": "Employee ID (e.g. emp_0001)"}, |
| "email": {"type": "string", "description": "Employee email address"}, |
| }, |
| }, |
| }, |
| { |
| "name": "hr_update_employee", |
| "description": "Update fields on an existing employee record. Cannot modify emp_id.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "emp_id": {"type": "string", "description": "Employee ID to update"}, |
| "updates": {"type": "object", "description": "Dictionary of fields to update (e.g. {status: 'active', department: 'Engineering'})"}, |
| }, |
| "required": ["emp_id", "updates"], |
| }, |
| }, |
| { |
| "name": "hr_search_employees", |
| "description": "Search employees by criteria. Filter by department, level, status, location, or role.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "department": {"type": "string", "description": "Filter by department"}, |
| "level": {"type": "string", "description": "Filter by level (L1-L6)"}, |
| "status": {"type": "string", "description": "Filter by status (active, pending, offboarded)"}, |
| "location": {"type": "string", "description": "Filter by location"}, |
| "role": {"type": "string", "description": "Filter by role/title"}, |
| "name": {"type": "string", "description": "Filter by name (exact match)"}, |
| }, |
| }, |
| }, |
| { |
| "name": "hr_get_org_chart", |
| "description": "Get the organizational hierarchy/reporting structure for a department. Shows manager-report relationships.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "department": {"type": "string", "description": "Department name"}, |
| }, |
| "required": ["department"], |
| }, |
| }, |
| { |
| "name": "onboarding_create_request", |
| "description": "Initiate an onboarding request for a new hire. The employee must exist in the HR system with 'pending' status. Creates a checklist of onboarding steps based on department.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "employee_id": {"type": "string", "description": "Employee ID of the new hire"}, |
| }, |
| "required": ["employee_id"], |
| }, |
| }, |
| { |
| "name": "onboarding_get_status", |
| "description": "Check the status of an onboarding request. Can look up by request ID or employee ID.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "request_id": {"type": "string", "description": "Onboarding request ID"}, |
| "employee_id": {"type": "string", "description": "Employee ID"}, |
| }, |
| }, |
| }, |
| { |
| "name": "onboarding_complete_step", |
| "description": "Mark an onboarding step as completed. Steps must be valid for the request.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "request_id": {"type": "string", "description": "Onboarding request ID"}, |
| "step": {"type": "string", "description": "Step name to mark as completed"}, |
| }, |
| "required": ["request_id", "step"], |
| }, |
| }, |
| { |
| "name": "offboarding_create_request", |
| "description": "Initiate an offboarding request for a departing employee. Specify reason (resignation, termination, transfer) and optional exit date.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "employee_id": {"type": "string", "description": "Employee ID"}, |
| "reason": {"type": "string", "description": "Reason for offboarding (resignation, termination, transfer)"}, |
| "exit_date": {"type": "string", "description": "Last working date (YYYY-MM-DD)"}, |
| }, |
| "required": ["employee_id"], |
| }, |
| }, |
| { |
| "name": "offboarding_get_status", |
| "description": "Check the status of an offboarding request. Can look up by request ID or employee ID.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "request_id": {"type": "string", "description": "Offboarding request ID"}, |
| "employee_id": {"type": "string", "description": "Employee ID"}, |
| }, |
| }, |
| }, |
| { |
| "name": "offboarding_complete_step", |
| "description": "Mark an offboarding step as completed.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "request_id": {"type": "string", "description": "Offboarding request ID"}, |
| "step": {"type": "string", "description": "Step name to mark as completed"}, |
| }, |
| "required": ["request_id", "step"], |
| }, |
| }, |
| { |
| "name": "it_assign_asset", |
| "description": "Assign an IT asset (laptop, monitor, phone, headset) to an employee. Asset must be available.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "asset_id": {"type": "string", "description": "Asset ID to assign"}, |
| "employee_id": {"type": "string", "description": "Employee ID to assign to"}, |
| }, |
| "required": ["asset_id", "employee_id"], |
| }, |
| }, |
| { |
| "name": "it_get_available_assets", |
| "description": "List available (unassigned) IT assets. Optionally filter by type.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "asset_type": {"type": "string", "description": "Filter by asset type (laptop, monitor, phone, headset)"}, |
| }, |
| }, |
| }, |
| { |
| "name": "it_create_account", |
| "description": "Create IT accounts (email, Slack, VPN, etc.) for an employee.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "employee_id": {"type": "string", "description": "Employee ID"}, |
| "account_types": { |
| "type": "array", |
| "items": {"type": "string"}, |
| "description": "Types of accounts to create (email, slack, vpn, github, jira, etc.)", |
| }, |
| }, |
| "required": ["employee_id"], |
| }, |
| }, |
| { |
| "name": "it_revoke_access", |
| "description": "Revoke all IT system access for an employee (used during offboarding).", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "employee_id": {"type": "string", "description": "Employee ID"}, |
| }, |
| "required": ["employee_id"], |
| }, |
| }, |
| { |
| "name": "it_get_software_licenses", |
| "description": "Check software license availability. Optionally filter by software name.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "software_name": {"type": "string", "description": "Filter by software name"}, |
| }, |
| }, |
| }, |
| { |
| "name": "access_assign_role", |
| "description": "Assign an access role to an employee. Checks level requirements and department restrictions.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "employee_id": {"type": "string", "description": "Employee ID"}, |
| "role_id": {"type": "string", "description": "Access role ID to assign"}, |
| }, |
| "required": ["employee_id", "role_id"], |
| }, |
| }, |
| { |
| "name": "access_create_badge", |
| "description": "Create a physical access badge for an employee. Server room access requires L4+ security approval.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "employee_id": {"type": "string", "description": "Employee ID"}, |
| "access_zones": { |
| "type": "array", |
| "items": {"type": "string"}, |
| "description": "List of access zones (main_entrance, floor_X, server_room, parking)", |
| }, |
| }, |
| "required": ["employee_id"], |
| }, |
| }, |
| { |
| "name": "access_revoke_role", |
| "description": "Revoke a specific access role from an employee.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "employee_id": {"type": "string", "description": "Employee ID"}, |
| "role_id": {"type": "string", "description": "Access role ID to revoke"}, |
| }, |
| "required": ["employee_id", "role_id"], |
| }, |
| }, |
| { |
| "name": "access_get_security_groups", |
| "description": "List all security groups. Optionally see members and resources.", |
| "parameters": { |
| "type": "object", |
| "properties": {}, |
| }, |
| }, |
| { |
| "name": "email_send", |
| "description": "Send an email. Used for welcome emails, notifications, farewell messages, etc.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "from_address": {"type": "string", "description": "Sender email address"}, |
| "to_address": {"type": "string", "description": "Recipient email address"}, |
| "subject": {"type": "string", "description": "Email subject line"}, |
| "body": {"type": "string", "description": "Email body text"}, |
| }, |
| "required": ["from_address", "to_address", "subject", "body"], |
| }, |
| }, |
| { |
| "name": "slack_send_message", |
| "description": "Post a message in a Slack channel or send a DM.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "channel": {"type": "string", "description": "Slack channel (e.g. #engineering, #general, @username)"}, |
| "sender": {"type": "string", "description": "Sender username or bot name"}, |
| "text": {"type": "string", "description": "Message text"}, |
| }, |
| "required": ["channel", "sender", "text"], |
| }, |
| }, |
| { |
| "name": "meeting_schedule", |
| "description": "Schedule a meeting (orientation, 1-on-1, exit interview, etc.).", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "title": {"type": "string", "description": "Meeting title"}, |
| "attendees": { |
| "type": "array", |
| "items": {"type": "string"}, |
| "description": "List of attendee employee IDs or emails", |
| }, |
| "datetime": {"type": "string", "description": "Meeting date and time (ISO format)"}, |
| "meeting_type": {"type": "string", "description": "Type of meeting (orientation, one_on_one, exit_interview, team_intro)"}, |
| }, |
| "required": ["title", "attendees", "datetime"], |
| }, |
| }, |
| { |
| "name": "policy_lookup", |
| "description": "Look up company policies by topic or department. Returns policy content and key rules.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "topic": {"type": "string", "description": "Policy topic to search for"}, |
| "department": {"type": "string", "description": "Department-specific policies"}, |
| "policy_id": {"type": "string", "description": "Specific policy ID"}, |
| }, |
| }, |
| }, |
| { |
| "name": "approval_request", |
| "description": "Submit an approval request (manager approval, IT approval, security approval). Approver must meet level requirements.", |
| "parameters": { |
| "type": "object", |
| "properties": { |
| "request_id": {"type": "string", "description": "The onboarding/offboarding request ID this approval is for"}, |
| "approver_id": {"type": "string", "description": "Employee ID of the approver"}, |
| "approval_type": {"type": "string", "description": "Type of approval (manager_approval, it_approval, security_approval, legal_approval)"}, |
| }, |
| "required": ["request_id", "approver_id", "approval_type"], |
| }, |
| }, |
| ] |
|
|
|
|
| class ToolRegistry: |
| """Registry of all available tools, mapping names to executors.""" |
|
|
| def __init__(self, world: WorldState): |
| self.world = world |
| self._tools: dict[str, Callable] = self._register_tools() |
|
|
| def _register_tools(self) -> dict[str, Callable]: |
| return { |
| "hr_create_employee": self._hr_create_employee, |
| "hr_read_employee": self._hr_read_employee, |
| "hr_update_employee": self._hr_update_employee, |
| "hr_search_employees": self._hr_search_employees, |
| "hr_get_org_chart": self._hr_get_org_chart, |
| "onboarding_create_request": self._onboarding_create_request, |
| "onboarding_get_status": self._onboarding_get_status, |
| "onboarding_complete_step": self._onboarding_complete_step, |
| "offboarding_create_request": self._offboarding_create_request, |
| "offboarding_get_status": self._offboarding_get_status, |
| "offboarding_complete_step": self._offboarding_complete_step, |
| "it_assign_asset": self._it_assign_asset, |
| "it_get_available_assets": self._it_get_available_assets, |
| "it_create_account": self._it_create_account, |
| "it_revoke_access": self._it_revoke_access, |
| "it_get_software_licenses": self._it_get_software_licenses, |
| "access_assign_role": self._access_assign_role, |
| "access_create_badge": self._access_create_badge, |
| "access_revoke_role": self._access_revoke_role, |
| "access_get_security_groups": self._access_get_security_groups, |
| "email_send": self._email_send, |
| "slack_send_message": self._slack_send_message, |
| "meeting_schedule": self._meeting_schedule, |
| "policy_lookup": self._policy_lookup, |
| "approval_request": self._approval_request, |
| } |
|
|
| @property |
| def tool_definitions(self) -> list[dict]: |
| return TOOL_DEFINITIONS |
|
|
| @property |
| def tool_names(self) -> list[str]: |
| return list(self._tools.keys()) |
|
|
| def execute(self, tool_name: str, params: dict) -> dict: |
| """Execute a tool by name with given parameters. Returns result dict.""" |
| if tool_name not in self._tools: |
| return {"success": False, "error": f"Unknown tool: {tool_name}. Available tools: {self.tool_names}"} |
| try: |
| result = self._tools[tool_name](params) |
| self.world.log_action(tool_name, params, result) |
| return result |
| except Exception as e: |
| error_result = {"success": False, "error": f"Tool execution error: {str(e)}"} |
| self.world.log_action(tool_name, params, error_result) |
| return error_result |
|
|
| |
|
|
| def _hr_create_employee(self, params: dict) -> dict: |
| return self.world.create_employee(params) |
|
|
| def _hr_read_employee(self, params: dict) -> dict: |
| emp = None |
| if "emp_id" in params: |
| emp = self.world.get_employee(params["emp_id"]) |
| elif "email" in params: |
| emp = self.world.get_employee_by_email(params["email"]) |
| if emp: |
| return {"success": True, "employee": emp} |
| return {"success": False, "error": "Employee not found"} |
|
|
| def _hr_update_employee(self, params: dict) -> dict: |
| return self.world.update_employee(params["emp_id"], params["updates"]) |
|
|
| def _hr_search_employees(self, params: dict) -> dict: |
| results = self.world.search_employees(**params) |
| return {"success": True, "count": len(results), "employees": results} |
|
|
| def _hr_get_org_chart(self, params: dict) -> dict: |
| chart = self.world.get_org_chart(params["department"]) |
| if chart: |
| return {"success": True, "org_chart": chart} |
| return {"success": False, "error": f"No employees found in department '{params['department']}'"} |
|
|
| def _onboarding_create_request(self, params: dict) -> dict: |
| return self.world.create_onboarding_request(params["employee_id"]) |
|
|
| def _onboarding_get_status(self, params: dict) -> dict: |
| status = self.world.get_onboarding_status( |
| request_id=params.get("request_id"), |
| emp_id=params.get("employee_id"), |
| ) |
| if status: |
| return {"success": True, "onboarding_status": status} |
| return {"success": False, "error": "Onboarding request not found"} |
|
|
| def _onboarding_complete_step(self, params: dict) -> dict: |
| return self.world.complete_onboarding_step(params["request_id"], params["step"]) |
|
|
| def _offboarding_create_request(self, params: dict) -> dict: |
| return self.world.create_offboarding_request( |
| params["employee_id"], |
| reason=params.get("reason", "resignation"), |
| exit_date=params.get("exit_date"), |
| ) |
|
|
| def _offboarding_get_status(self, params: dict) -> dict: |
| status = self.world.get_offboarding_status( |
| request_id=params.get("request_id"), |
| emp_id=params.get("employee_id"), |
| ) |
| if status: |
| return {"success": True, "offboarding_status": status} |
| return {"success": False, "error": "Offboarding request not found"} |
|
|
| def _offboarding_complete_step(self, params: dict) -> dict: |
| return self.world.complete_offboarding_step(params["request_id"], params["step"]) |
|
|
| def _it_assign_asset(self, params: dict) -> dict: |
| return self.world.assign_asset(params["asset_id"], params["employee_id"]) |
|
|
| def _it_get_available_assets(self, params: dict) -> dict: |
| assets = self.world.get_available_assets(params.get("asset_type")) |
| return {"success": True, "count": len(assets), "assets": assets} |
|
|
| def _it_create_account(self, params: dict) -> dict: |
| return self.world.create_it_account( |
| params["employee_id"], |
| account_types=params.get("account_types"), |
| ) |
|
|
| def _it_revoke_access(self, params: dict) -> dict: |
| return self.world.revoke_it_access(params["employee_id"]) |
|
|
| def _it_get_software_licenses(self, params: dict) -> dict: |
| licenses = self.world.get_software_licenses(params.get("software_name")) |
| return {"success": True, "licenses": licenses} |
|
|
| def _access_assign_role(self, params: dict) -> dict: |
| return self.world.assign_role(params["employee_id"], params["role_id"]) |
|
|
| def _access_create_badge(self, params: dict) -> dict: |
| return self.world.create_badge( |
| params["employee_id"], |
| access_zones=params.get("access_zones"), |
| ) |
|
|
| def _access_revoke_role(self, params: dict) -> dict: |
| return self.world.revoke_role(params["employee_id"], params["role_id"]) |
|
|
| def _access_get_security_groups(self, params: dict) -> dict: |
| return {"success": True, "security_groups": self.world.state["security_groups"]} |
|
|
| def _email_send(self, params: dict) -> dict: |
| return self.world.send_email( |
| params["from_address"], params["to_address"], |
| params["subject"], params["body"], |
| ) |
|
|
| def _slack_send_message(self, params: dict) -> dict: |
| return self.world.send_slack_message( |
| params["channel"], params["sender"], params["text"], |
| ) |
|
|
| def _meeting_schedule(self, params: dict) -> dict: |
| return self.world.schedule_meeting( |
| params["title"], params["attendees"], |
| params["datetime"], params.get("meeting_type", "general"), |
| ) |
|
|
| def _policy_lookup(self, params: dict) -> dict: |
| policies = self.world.lookup_policy( |
| topic=params.get("topic"), |
| department=params.get("department"), |
| policy_id=params.get("policy_id"), |
| ) |
| return {"success": True, "count": len(policies), "policies": policies} |
|
|
| def _approval_request(self, params: dict) -> dict: |
| return self.world.create_approval( |
| params["request_id"], params["approver_id"], params["approval_type"], |
| ) |
|
|
