simple_server.py•3.21 kB
#!/usr/bin/env python3
"""Minimal MCP Server - Single Tool Example"""
import asyncio
import sys
from pathlib import Path
# Add parent to path
sys.path.insert(0, str(Path(__file__).parent))
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from src.database import Database, CustomerDB
# Determine project root
PROJECT_ROOT = Path(__file__).parent
DATA_DIR = PROJECT_ROOT / "data"
async def main():
# Initialize database
db = Database(str(DATA_DIR / "travel_company.db"))
customer_db = CustomerDB(db.conn)
# Create server
server = Server("simple-travel-mcp")
# Register list_tools handler
@server.list_tools()
async def list_tools() -> list[Tool]:
return [
Tool(
name="search_customers",
description="Search for customers by name, email, phone, or customer ID",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query"
},
"search_by": {
"type": "string",
"description": "Field to search by",
"enum": ["name", "email", "phone", "customer_id"],
"default": "name"
}
},
"required": ["query"]
}
)
]
# Register call_tool handler
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
if name == "search_customers":
query = arguments.get("query", "")
search_by = arguments.get("search_by", "name")
if not query:
return [TextContent(type="text", text="Error: query parameter is required")]
# Call database
results = customer_db.search(query, search_by)
if not results:
return [TextContent(type="text", text=f"No customers found matching '{query}' in {search_by}")]
# Format output
output = f"Found {len(results)} customer(s):\n\n"
for customer in results:
output += f"Customer ID: {customer['customer_id']}\n"
output += f"Name: {customer['name']}\n"
output += f"Email: {customer['email']}\n"
output += f"Phone: {customer['phone']}\n"
output += f"Location: {customer['city']}, {customer['state']}\n"
output += f"Loyalty Tier: {customer['loyalty_tier']}\n"
output += "-" * 50 + "\n"
return [TextContent(type="text", text=output)]
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
# Run server
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())