feat: add staff availability and allocation tool

- Add 'check_staff_availability' tool to MCP server
- Implements cross-referencing /v1/employee and /v1/allocation_reporting
- Calculates availability percentage (100% - allocated)
- Supports filtering by skill and minimum availability threshold

Author: failattu

src/agileday_server.py

@@ -3,6 +3,7 @@
 import argparse
 import requests
 import uvicorn
+from datetime import date, timedelta
 from starlette.applications import Starlette
 from starlette.routing import Route
 from starlette.responses import Response
@@ -58,13 +59,27 @@ async def list_tools() -> list[Tool]:
                 "required": ["name_query"]
             }
         ),
-        # NEW TOOL ADDED HERE
         Tool(
             name="list_organization_skills",
             description="Retrieves a full list of all available skills/competences defined in the organization.",
             inputSchema={
                 "type": "object",
-                "properties": {}, # No arguments needed
+                "properties": {}, 
+            }
+        ),
+        # NEW TOOL: Availability Checker
+        Tool(
+            name="check_staff_availability",
+            description="Checks who is free and views current allocations for a date range. Can filter by skill.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "start_date": {"type": "string", "description": "Start date in YYYY-MM-DD format"},
+                    "end_date": {"type": "string", "description": "End date in YYYY-MM-DD format"},
+                    "skill_filter": {"type": "string", "description": "Optional: Filter for employees with this skill (e.g. 'React')"},
+                    "min_availability": {"type": "integer", "description": "Optional: Minimum free percentage to show (0-100). Default 0."}
+                },
+                "required": ["start_date", "end_date"]
             }
         )
     ]
@@ -75,10 +90,16 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent | ImageConte
         return [TextContent(type="text", text=find_experts_logic(arguments["skill_name"]))]
     if name == "get_employee_competence_profile":
         return [TextContent(type="text", text=profile_logic(arguments["name_query"]))]
-    
-    # NEW LOGIC HANDLER
     if name == "list_organization_skills":
         return [TextContent(type="text", text=list_skills_logic())]
+    # NEW HANDLER
+    if name == "check_staff_availability":
+        return [TextContent(type="text", text=check_availability_logic(
+            arguments["start_date"], 
+            arguments["end_date"], 
+            arguments.get("skill_filter"),
+            arguments.get("min_availability", 0)
+        ))]
 
     raise ValueError(f"Unknown tool: {name}")
 
@@ -138,7 +159,6 @@ def profile_logic(name_query: str) -> str:
 
     return "\n\n".join(results) if results else f"No employee found for '{name_query}'."
 
-# NEW LOGIC FUNCTION
 def list_skills_logic() -> str:
     try:
         url = f"{BASE_URL}/v1/skill"
@@ -151,24 +171,114 @@ def list_skills_logic() -> str:
     if not skills:
         return "No skills defined in the library."
 
-    # Group skills by category for better readability
     categories = {}
     for s in skills:
-        # Default to "Other" if category is missing/null
         cat = s.get('category') or "Uncategorized"
-        if cat not in categories: 
-            categories[cat] = []
+        if cat not in categories: categories[cat] = []
         categories[cat].append(s.get('name', 'Unnamed Skill'))
 
     output = ["**Organization Skill Library**\n"]
-    
-    # Sort categories alphabetically
     for cat in sorted(categories.keys()):
         skill_names = sorted(categories[cat])
-        # Format: "### Category Name \n - Skill 1, Skill 2..."
         output.append(f"### {cat}")
         output.append(", ".join(skill_names))
-        output.append("") # Empty line for spacing
+        output.append("")
+
+    return "\n".join(output)
+
+def check_availability_logic(start_date: str, end_date: str, skill_filter: str = None, min_availability: int = 0) -> str:
+    # 1. Fetch all employees first
+    try:
+        emp_url = f"{BASE_URL}/v1/employee"
+        emp_response = requests.get(emp_url, headers=get_headers(), timeout=15)
+        emp_response.raise_for_status()
+        all_employees = emp_response.json()
+    except Exception as e:
+        return f"Error fetching employees: {str(e)}"
+
+    # 2. Filter employees if skill_filter is active
+    relevant_employee_ids = set()
+    relevant_employees = {}
+    
+    for emp in all_employees:
+        emp_id = emp.get('id')
+        if not emp_id: continue
+        
+        # If skill filter applies, check skills
+        if skill_filter:
+            skills = emp.get('skills', [])
+            has_skill = any(skill_filter.lower() in s.get('name', '').lower() for s in skills)
+            if not has_skill:
+                continue
+        
+        relevant_employee_ids.add(emp_id)
+        relevant_employees[emp_id] = emp
+
+    if not relevant_employee_ids:
+        return f"No employees found matching skill: {skill_filter}" if skill_filter else "No employees found."
+
+    # 3. Fetch allocations for the period
+    try:
+        alloc_url = f"{BASE_URL}/v1/allocation_reporting"
+        params = {"startDate": start_date, "endDate": end_date}
+        alloc_response = requests.get(alloc_url, headers=get_headers(), params=params, timeout=15)
+        alloc_response.raise_for_status()
+        allocations = alloc_response.json()
+    except Exception as e:
+        return f"Error fetching allocations: {str(e)}"
+
+    # 4. Aggregate allocations by employee
+    # Map: employee_id -> list of project allocations
+    emp_workload = {eid: [] for eid in relevant_employee_ids}
+    emp_total_alloc = {eid: 0.0 for eid in relevant_employee_ids}
+
+    for alloc in allocations:
+        eid = alloc.get('employeeId')
+        # Only process if this employee is in our filtered list
+        if eid in relevant_employee_ids:
+            # According to schema, 'allocation' is the percentage
+            percentage = alloc.get('allocation', 0)
+            project_name = alloc.get('projectName', 'Unknown Project')
+            
+            emp_total_alloc[eid] += percentage
+            emp_workload[eid].append(f"{project_name} ({percentage}%)")
+
+    # 5. Format Output
+    output = []
+    output.append(f"**Availability Report ({start_date} to {end_date})**")
+    if skill_filter:
+        output.append(f"Filter: Skill matching '{skill_filter}'\n")
+
+    # Sort by availability (most free first)
+    sorted_eids = sorted(relevant_employee_ids, key=lambda x: emp_total_alloc[x])
+
+    for eid in sorted_eids:
+        total_load = emp_total_alloc[eid]
+        availability = 100 - total_load
+        
+        # Skip if below requested availability
+        if availability < min_availability:
+            continue
+
+        emp = relevant_employees[eid]
+        full_name = f"{emp.get('firstName', '')} {emp.get('lastName', '')}".strip()
+        title = emp.get('title', 'No Title')
+
+        # Status Icon
+        if availability >= 80:
+            icon = "🟢" # Mostly free
+        elif availability >= 20:
+            icon = "🟡" # Partially booked
+        else:
+            icon = "🔴" # Busy
+
+        # Work details
+        work_desc = ", ".join(emp_workload[eid]) if emp_workload[eid] else "No allocations"
+        
+        output.append(f"{icon} **{full_name}** ({title})")
+        output.append(f"   Availability: **{availability:.0f}%** (Allocated: {total_load:.0f}%)")
+        output.append(f"   Current Work: {work_desc}")
+        output.append("")
 
     return "\n".join(output)
 

tests/test_agileday_server.py

@@ -11,7 +11,7 @@
 sys.path.append(os.path.join(os.path.dirname(__file__), '../src'))
 
 # Add list_skills_logic to imports
-from agileday_server import find_experts_logic, profile_logic, list_skills_logic, BASE_URL
+from agileday_server import find_experts_logic, profile_logic, list_skills_logic, check_availability_logic, BASE_URL
 
 # --- Fixtures ---
 
@@ -131,4 +131,47 @@ def test_list_skills_grouping(mock_api_response):
     assert "### Frontend" in result
     assert "React" in result
     assert "### Uncategorized" in result
-    assert "Figma" in result
+    assert "Figma" in result
+
+def test_check_availability(mock_api_response):
+    """Test the availability calculation logic."""
+    # 1. Setup Mock for Employees (First call)
+    employees_data = [
+        {"id": "user1", "firstName": "Alice", "lastName": "Free", "title": "Dev"},
+        {"id": "user2", "firstName": "Bob", "lastName": "Busy", "title": "Lead"}
+    ]
+    
+    # 2. Setup Mock for Allocations (Second call)
+    allocations_data = [
+        # Bob is busy with 100% allocation
+        {"employeeId": "user2", "projectName": "Project X", "allocation": 100}
+    ]
+
+    # Configure side_effect to return different data for different URLs
+    def side_effect(*args, **kwargs):
+        if "employee" in args[0]:
+            mock = MagicMock()
+            mock.status_code = 200
+            mock.json.return_value = employees_data
+            return mock
+        if "allocation_reporting" in args[0]:
+            mock = MagicMock()
+            mock.status_code = 200
+            mock.json.return_value = allocations_data
+            return mock
+        return MagicMock()
+
+    mock_api_response.side_effect = side_effect
+
+    # 3. Run Logic
+    result = check_availability_logic("2024-01-01", "2024-01-31")
+
+    # 4. Assertions
+    # Alice should be green (100% free)
+    assert "🟢 **Alice Free**" in result
+    assert "Availability: **100%**" in result
+    
+    # Bob should be red (0% free)
+    assert "🔴 **Bob Busy**" in result
+    assert "Allocated: 100%" in result
+    assert "Project X" in result