MCP Server Setup

curl -LsSf https://astral.sh/uv/install.sh | sh

# Create a new directory for our project
uv init employee-server
cd employee-server

# Create virtual environment and activate it
uv venv
source .venv/bin/activate

uv add fastmcp

from fastmcp import FastMCP
from typing import Dict, Optional

# Initialize FastMCP server
mcp = FastMCP("Employee Knowledge Base")

# Sample employee data (in production, this would come from a database)
EMPLOYEE_DATA = {
    "EMP001": {
        "name": "Alice Johnson",
        "department": "Engineering",
        "role": "Senior Developer",
        "salary": 120000,
        "email": "alice.johnson@company.com",
        "manager": "EMP005"
    },
    "EMP002": {
        "name": "Bob Smith",
        "department": "Sales",
        "role": "Sales Manager",
        "salary": 95000,
        "email": "bob.smith@company.com",
        "manager": "EMP006"
    },
    "EMP003": {
        "name": "Carol White",
        "department": "HR",
        "role": "HR Specialist",
        "salary": 75000,
        "email": "carol.white@company.com",
        "manager": "EMP007"
    },
    "EMP004": {
        "name": "David Brown",
        "department": "Engineering",
        "role": "Junior Developer",
        "salary": 80000,
        "email": "david.brown@company.com",
        "manager": "EMP005"
    },
    "EMP005": {
        "name": "Eva Martinez",
        "department": "Engineering",
        "role": "Engineering Manager",
        "salary": 150000,
        "email": "eva.martinez@company.com",
        "manager": "EMP008"
    }
}

@mcp.tool()
def get_employee_by_id(employee_id: str) -> Dict:
    """
    Retrieve employee information by their ID.
    
    Args:
        employee_id: The unique employee identifier (e.g., EMP001)
    
    Returns:
        Employee information including name, department, role, and salary
    """
    if employee_id in EMPLOYEE_DATA:
        return {
            "success": True,
            "data": EMPLOYEE_DATA[employee_id]
        }
    else:
        return {
            "success": False,
            "error": f"Employee with ID {employee_id} not found"
        }

@mcp.tool()
def search_employees_by_department(department: str) -> Dict:
    """
    Search for all employees in a specific department.
    
    Args:
        department: The department name to search for
    
    Returns:
        List of employees in the specified department
    """
    employees = []
    for emp_id, emp_data in EMPLOYEE_DATA.items():
        if emp_data["department"].lower() == department.lower():
            employees.append({
                "id": emp_id,
                **emp_data
            })
    
    return {
        "success": True,
        "count": len(employees),
        "employees": employees
    }

@mcp.tool()
def get_salary_range(min_salary: Optional[int] = None, max_salary: Optional[int] = None) -> Dict:
    """
    Find employees within a specific salary range.
    
    Args:
        min_salary: Minimum salary threshold (optional)
        max_salary: Maximum salary threshold (optional)
    
    Returns:
        List of employees within the specified salary range
    """
    employees = []
    for emp_id, emp_data in EMPLOYEE_DATA.items():
        salary = emp_data["salary"]
        if (min_salary is None or salary >= min_salary) and \
           (max_salary is None or salary <= max_salary):
            employees.append({
                "id": emp_id,
                "name": emp_data["name"],
                "department": emp_data["department"],
                "salary": salary
            })
    
    # Sort by salary
    employees.sort(key=lambda x: x["salary"], reverse=True)
    
    return {
        "success": True,
        "count": len(employees),
        "employees": employees
    }

@mcp.tool()
def get_employee_hierarchy(employee_id: str) -> Dict:
    """
    Get the reporting hierarchy for an employee.
    
    Args:
        employee_id: The employee ID to get hierarchy for
    
    Returns:
        The employee's manager and any direct reports
    """
    if employee_id not in EMPLOYEE_DATA:
        return {
            "success": False,
            "error": f"Employee with ID {employee_id} not found"
        }
    
    employee = EMPLOYEE_DATA[employee_id]
    
    # Find manager
    manager = None
    if employee.get("manager") and employee["manager"] in EMPLOYEE_DATA:
        manager = {
            "id": employee["manager"],
            "name": EMPLOYEE_DATA[employee["manager"]]["name"],
            "role": EMPLOYEE_DATA[employee["manager"]]["role"]
        }
    
    # Find direct reports
    direct_reports = []
    for emp_id, emp_data in EMPLOYEE_DATA.items():
        if emp_data.get("manager") == employee_id:
            direct_reports.append({
                "id": emp_id,
                "name": emp_data["name"],
                "role": emp_data["role"]
            })
    
    return {
        "success": True,
        "employee": {
            "id": employee_id,
            "name": employee["name"],
            "role": employee["role"]
        },
        "manager": manager,
        "direct_reports": direct_reports
    }

@mcp.tool()
def get_department_statistics() -> Dict:
    """
    Get statistics for all departments including employee count and average salary.
    
    Returns:
        Statistics for each department
    """
    dept_stats = {}
    
    for emp_data in EMPLOYEE_DATA.values():
        dept = emp_data["department"]
        if dept not in dept_stats:
            dept_stats[dept] = {
                "count": 0,
                "total_salary": 0,
                "employees": []
            }
        
        dept_stats[dept]["count"] += 1
        dept_stats[dept]["total_salary"] += emp_data["salary"]
        dept_stats[dept]["employees"].append(emp_data["name"])
    
    # Calculate averages
    result = {}
    for dept, stats in dept_stats.items():
        result[dept] = {
            "employee_count": stats["count"],
            "average_salary": stats["total_salary"] / stats["count"],
            "total_salary": stats["total_salary"],
            "employees": stats["employees"]
        }
    
    return {
        "success": True,
        "departments": result
    }

# Run the server
if __name__ == "__main__":
    mcp.run(transport="streamable-http", path="/mcp", port=8000)

python employee_server.py

{"jsonrpc":"2.0","id":"server-error","error":{"code":-32600,"message":"Not Acceptable: Client must accept text/event-stream"}}

npx @modelcontextprotocol/inspector@latest

⚙️ Proxy server listening on localhost:6277
🔑 Session token: <session-token>
   Use this token to authenticate requests or set DANGEROUSLY_OMIT_AUTH=true to disable auth

🚀 MCP Inspector is up and running at:
   <http://localhost:6274/?MCP_PROXY_AUTH_TOKEN=><session-token>

ngrok http 8000

{"jsonrpc":"2.0","id":"server-error","error":{"code":-32600,"message":"Not Acceptable: Client must accept text/event-stream"}}

from ai21 import AsyncAI21Client, version

client = AsyncAI21Client(api_key="ai21-api-key")

run = await client.beta.maestro.runs.create_and_poll(
    input="What is Alice's role and to which department does she belong?",
    tools=[
        {
            "type": "mcp",
            "server_url": "https://abc123.ngrok-free.app/mcp",
            "server_label": "Employees",
        },
    ],
    budget="medium",
)

print("id:", run.id)
print("Result:", run.result)

id: 068c664b-f9ea-7078-8000-be3f80ffa83c
Result: Eva Martinez, who is an Engineering Manager, has the following direct reports:

1. Alice Johnson - Senior Developer
2. David Brown - Junior Developer