Building an Employee Server with FastMCP
This guide walks you through creating a Python-based MCP server that represents an employee management system. Using AI21 Maestro, you will be able to ask questions about your departments and employees. We’ll expose it remotely with ngrok and call it using AI21 Maestro.Prerequisites
- Python 3.10+
- ngrok account (free tier works)
- AI21 Maestro platform access
Step 1: Create the MCP Server
Show Content
Show Content
1.1 Install uv and initialize the project
uv
is a modern, extremely fast Python package and project manager.First, let’s install
uv
and set up our Python project and environment:Copy
Ask AI
curl -LsSf https://astral.sh/uv/install.sh | sh
Copy
Ask AI
# 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
Copy
Ask AI
uv add fastmcp
1.2 Create the Server File
In theemployee-server
directory, create a file named employee_server.py
:Copy
Ask AI
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_employee_by_name(name_query: str, max_results: Optional[int] = 10) -> Dict:
"""
Search for employees by name using fuzzy matching.
Args:
name_query: The name or partial name to search for
max_results: Maximum number of results to return (default: 10)
Returns:
List of employees matching the name query, sorted by relevance
"""
if not name_query.strip():
return {
"success": False,
"error": "Name query cannot be empty"
}
query = name_query.lower().strip()
matches = []
for emp_id, emp_data in EMPLOYEE_DATA.items():
employee_name = emp_data["name"].lower()
score = 0
# Exact match gets highest score
if query == employee_name:
score = 100
# Check if query is contained in name
elif query in employee_name:
score = 80
# Check if all words in query are in name
elif all(word in employee_name for word in query.split()):
score = 60
# Check if any word in query matches any word in name
elif any(word in employee_name for word in query.split()):
score = 40
# Check if name starts with query
elif employee_name.startswith(query):
score = 70
# Check for partial word matches
else:
query_words = query.split()
name_words = employee_name.split()
partial_matches = 0
for q_word in query_words:
for n_word in name_words:
if q_word in n_word or n_word in q_word:
partial_matches += 1
break
if partial_matches > 0:
score = 20 + (partial_matches * 10)
if score > 0:
matches.append({
"id": emp_id,
"score": score
})
# Sort by score (highest first)
matches.sort(key=lambda x: x["score"], reverse=True)
# Limit results
if max_results:
matches = matches[:max_results]
return {
"success": True,
"query": name_query,
"count": len(matches),
"employee_ids": [match["id"] for match in matches]
}
@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)
1.3 Run the Local Server in development mode
Copy
Ask AI
python employee_server.py
Copy
Ask AI
{"jsonrpc":"2.0","id":"server-error","error":{"code":-32600,"message":"Not Acceptable: Client must accept text/event-stream"}}
Step 2: Test the MCP Server
Show Content
Show Content
Now we will test that the server is running and that it adheres to the Model Context Protocol correctly by listing its tools.We will use the MCP Inspector tool which creates a locally running MCP client along with a user facing web interface.You should see the following output:Copy the session token from this output.
2.1 Run the MCP Inspector
In a different terminal run the following command:Copy
Ask AI
npx @modelcontextprotocol/inspector@latest
Copy
Ask AI
⚙️ 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>
2.2 List your server tools
Open the browser inhttp://localhost:6274/
and you should see the Inspector UI.- Choose Streamable HTTP as the Transport Type
- Set the URL to
http://localhost:8000/mcp
- Go to the Configuration pane and paste the
<session-token>
in the Proxy Session Token textbox - Click on the Connect button
- Select Tools and then click on “List Tools”
- You should see a list of all the tools your server exposes

- Optional: you can interact with each tool by clicking its name in the list and providing the required parameters
Step 3: Expose Server with ngrok
Show Content
Show Content
2.1 Install ngrok
Download ngrok from ngrok.com and create a free account.2.2 Start ngrok Tunnel
Copy
Ask AI
ngrok http 8000
https://abc123.ngrok-free.app
Important: Save this URL, you’ll need it for AI21 Maestro integration.Test the URL by making a request to https://abc123.ngrok-free.app/mcp
. and getting You should get the following response:Copy
Ask AI
{"jsonrpc":"2.0","id":"server-error","error":{"code":-32600,"message":"Not Acceptable: Client must accept text/event-stream"}}
Step 4: Integrate with AI21 Maestro
Show Content
Show Content
4.1 Run with AI21 Maestro
In your AI21 Maestro configuration, add the remote MCP server:Copy
Ask AI
import asyncio
from ai21 import AsyncAI21Client
client = AsyncAI21Client(api_key="<your-api-key>")
async def main():
run = await client.beta.maestro.runs.create_and_poll(
input="Who reports to Eva Martinez?",
tools=[
{
"type": "mcp",
"server_url": "https://<app-id>.ngrok-free.app/mcp",
"server_label": "Employees",
},
],
budget="medium",
)
print("id:", run.id)
print("Status:", run.status)
print("Result:", run.result)
# Works in Jupyter
await main()
# Comment the above and uncomment the below to run in Python scripts
# import asyncio
# asyncio.run(main())
Copy
Ask AI
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
4.2 Test in AI21 Maestro
Example queries to test in AI21 Maestro:- “What is the salary of employee EMP001?”
- “Show me all employees in the Engineering department”
- “Find employees earning between 80000 and 120000”
- “Who reports to EMP005?”
- “What are the department statistics?”