"""Main MCP server setup and tool registration."""
import asyncio
import logging
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from .config import config
from .domains.career import read as career_read
from .domains.career import write as career_write
from .domains.career import analysis as career_analysis
from .utils.storage import init_database
from .utils.logging import setup_logging
# Initialize logging
setup_logging(config.log_level)
logger = logging.getLogger(__name__)
# Create MCP server instance
app = Server("personal-productivity-mcp")
# ============================================================================
# Tool Registration
# ============================================================================
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List all available MCP tools."""
return [
# Career Read Tools
Tool(
name="career_search_greenhouse",
description="Search for jobs on a company's Greenhouse job board",
inputSchema={
"type": "object",
"properties": {
"company_name": {
"type": "string",
"description": "Company name to search for jobs"
}
},
"required": ["company_name"]
}
),
Tool(
name="career_search_lever",
description="Search for jobs on a company's Lever job board",
inputSchema={
"type": "object",
"properties": {
"company_name": {
"type": "string",
"description": "Company name to search for jobs"
}
},
"required": ["company_name"]
}
),
Tool(
name="career_get_applications",
description="Get tracked job applications, optionally filtered by status",
inputSchema={
"type": "object",
"properties": {
"status": {
"type": "string",
"description": "Filter by status: interested, applied, interviewing, offered, rejected",
"enum": ["interested", "applied", "interviewing", "offered", "rejected"]
}
}
}
),
# Career Write Tools
Tool(
name="career_track_application",
description="Track a new job application",
inputSchema={
"type": "object",
"properties": {
"job_url": {
"type": "string",
"description": "URL to the job posting"
},
"company": {
"type": "string",
"description": "Company name"
},
"title": {
"type": "string",
"description": "Job title"
},
"status": {
"type": "string",
"description": "Application status",
"enum": ["interested", "applied", "interviewing", "offered", "rejected"]
},
"notes": {
"type": "string",
"description": "Notes about the application"
}
},
"required": ["job_url", "company", "title", "status"]
}
),
Tool(
name="career_update_application",
description="Update an existing job application status",
inputSchema={
"type": "object",
"properties": {
"app_id": {
"type": "string",
"description": "Application ID to update"
},
"status": {
"type": "string",
"description": "New application status",
"enum": ["interested", "applied", "interviewing", "offered", "rejected"]
},
"notes": {
"type": "string",
"description": "Additional notes"
}
},
"required": ["app_id", "status"]
}
),
# Career Analysis Tools
Tool(
name="career_analyze_pipeline",
description="Analyze the job application pipeline and get statistics",
inputSchema={
"type": "object",
"properties": {}
}
),
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""Route tool calls to appropriate handlers."""
logger.info(f"Tool called: {name} with arguments: {arguments}")
try:
# Career Read Tools
if name == "career_search_greenhouse":
result = await career_read.search_greenhouse_boards(
arguments["company_name"]
)
return [TextContent(type="text", text=str(result))]
elif name == "career_search_lever":
result = await career_read.search_lever_boards(
arguments["company_name"]
)
return [TextContent(type="text", text=str(result))]
elif name == "career_get_applications":
result = await career_read.get_applications(
arguments.get("status")
)
return [TextContent(type="text", text=str(result))]
# Career Write Tools
elif name == "career_track_application":
result = await career_write.track_application(
job_url=arguments["job_url"],
company=arguments["company"],
title=arguments["title"],
status=arguments["status"],
notes=arguments.get("notes", "")
)
return [TextContent(type="text", text=str(result))]
elif name == "career_update_application":
result = await career_write.update_application_status(
app_id=arguments["app_id"],
status=arguments["status"],
notes=arguments.get("notes", "")
)
return [TextContent(type="text", text=str(result))]
# Career Analysis Tools
elif name == "career_analyze_pipeline":
result = await career_analysis.analyze_application_pipeline()
return [TextContent(type="text", text=str(result))]
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
logger.error(f"Error executing tool {name}: {e}", exc_info=True)
return [TextContent(
type="text",
text=f"Error executing {name}: {str(e)}"
)]
# ============================================================================
# Server Lifecycle
# ============================================================================
async def main():
"""Run the MCP server."""
logger.info("Starting Personal Productivity MCP Server")
logger.info(f"Server version: 0.1.0")
logger.info(f"Database path: {config.database_path}")
# Initialize database
await init_database()
logger.info("Database initialized")
# Start server with stdio transport
async with stdio_server() as (read_stream, write_stream):
logger.info("Server running on stdio")
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
