mcp_resume_server_http.py•9.46 kB
#!/usr/bin/env python3
"""
Personal Resume MCP Server - HTTP Version for AWS Deployment
HTTP-based MCP server for cloud deployment and remote access
"""
import asyncio
import json
import logging
import os
from typing import Dict, Any, Optional
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from personal_resume_agent import PersonalResumeAgent
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(
title="Personal Resume MCP Server",
description="MCP server for personal resume queries via HTTP",
version="1.0.0"
)
# Enable CORS for Claude Desktop integration
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Configure appropriately for production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
class PersonalResumeMCPHTTPServer:
"""HTTP-based MCP Server for Personal Resume Agent"""
def __init__(self):
"""Initialize the HTTP MCP server"""
self.agent = PersonalResumeAgent()
self.initialized = False
async def initialize(self) -> bool:
"""Initialize the server and agent"""
try:
success = await self.agent.initialize()
if success:
self.initialized = True
logger.info("✅ Personal Resume MCP HTTP Server initialized")
return success
except Exception as e:
logger.error(f"Failed to initialize MCP server: {e}")
return False
def create_response(self, request_id: Optional[str], result: Optional[Dict[str, Any]] = None, error: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Create a properly formatted JSON-RPC response"""
response = {
"jsonrpc": "2.0",
"id": request_id if request_id is not None else 0
}
if error:
response["error"] = error
else:
response["result"] = result if result is not None else {}
return response
async def handle_initialize(self, request_id: Optional[str]) -> Dict[str, Any]:
"""Handle MCP initialization"""
if not self.initialized:
await self.initialize()
return self.create_response(request_id, {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"serverInfo": {
"name": "personal-resume-server-http",
"version": "1.0.0"
}
})
async def handle_tools_list(self, request_id: Optional[str]) -> Dict[str, Any]:
"""Handle tool listing request"""
tools = [
{
"name": "query_resume",
"description": "Query personal resume information",
"inputSchema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Question about resume"
}
},
"required": ["query"]
}
},
{
"name": "get_agent_info",
"description": "Get agent capabilities and status",
"inputSchema": {
"type": "object",
"properties": {}
}
},
{
"name": "analyze_skill_match",
"description": "Compare skills with job requirements",
"inputSchema": {
"type": "object",
"properties": {
"job_requirements": {
"type": "string",
"description": "Job description or requirements"
}
},
"required": ["job_requirements"]
}
}
]
return self.create_response(request_id, {"tools": tools})
async def handle_tool_call(self, request_id: Optional[str], params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle tool call requests"""
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name == "query_resume":
query = arguments.get("query", "")
if not query:
return self.create_response(request_id, error={
"code": -32602,
"message": "Missing required parameter: query"
})
result = await self.agent.process_query(query)
response_text = result.get('response', 'No response available')
return self.create_response(request_id, {
"content": [
{
"type": "text",
"text": response_text
}
]
})
elif tool_name == "get_agent_info":
info = self.agent.get_agent_info()
return self.create_response(request_id, {
"content": [
{
"type": "text",
"text": f"Agent Status: {info['agent_name']}\n"
f"Initialized: {info['initialized']}\n"
f"Capabilities: {', '.join(info['capabilities'])}\n"
f"Resume Summary: {info['resume_summary']}"
}
]
})
elif tool_name == "analyze_skill_match":
job_requirements = arguments.get("job_requirements", "")
if not job_requirements:
return self.create_response(request_id, error={
"code": -32602,
"message": "Missing required parameter: job_requirements"
})
match_result = await self.agent.get_skill_match(job_requirements)
response_text = f"Skill Match Analysis:\n"
response_text += f"Match Percentage: {match_result.get('match_percentage', 0)}%\n"
response_text += f"Matching Skills: {', '.join(match_result.get('matching_skills', []))}\n"
response_text += f"Confidence: {match_result.get('confidence', 0):.2f}"
return self.create_response(request_id, {
"content": [
{
"type": "text",
"text": response_text
}
]
})
return self.create_response(request_id, error={
"code": -32602,
"message": f"Unknown tool: {tool_name}"
})
# Global server instance
mcp_server = PersonalResumeMCPHTTPServer()
@app.on_event("startup")
async def startup_event():
"""Initialize the MCP server on startup"""
await mcp_server.initialize()
@app.post("/mcp")
async def mcp_endpoint(request: Request):
"""Main MCP endpoint for JSON-RPC requests"""
try:
body = await request.json()
method = body.get("method")
request_id = body.get("id")
params = body.get("params", {})
# Handle request
if method == "initialize":
response = await mcp_server.handle_initialize(request_id)
elif method == "tools/list":
response = await mcp_server.handle_tools_list(request_id)
elif method == "tools/call":
response = await mcp_server.handle_tool_call(request_id, params)
elif method == "prompts/list":
response = mcp_server.create_response(request_id, {"prompts": []})
elif method == "resources/list":
response = mcp_server.create_response(request_id, {"resources": []})
else:
response = mcp_server.create_response(request_id, error={
"code": -32601,
"message": f"Method not found: {method}"
})
return response
except json.JSONDecodeError as e:
logger.error(f"Invalid JSON: {e}")
raise HTTPException(status_code=400, detail="Invalid JSON")
except Exception as e:
logger.error(f"Error processing request: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"initialized": mcp_server.initialized,
"agent_ready": mcp_server.agent.initialized if hasattr(mcp_server.agent, 'initialized') else False
}
@app.get("/")
async def root():
"""Root endpoint with server info"""
return {
"name": "Personal Resume MCP Server",
"version": "1.0.0",
"description": "HTTP-based MCP server for personal resume queries",
"endpoints": {
"mcp": "/mcp (POST) - Main MCP JSON-RPC endpoint",
"health": "/health (GET) - Health check",
"docs": "/docs (GET) - API documentation"
}
}
def main():
"""Main entry point"""
port = int(os.environ.get("PORT", 8000))
logger.info(f"🚀 Starting Personal Resume MCP HTTP Server on port {port}")
uvicorn.run(
"mcp_resume_server_http:app",
host="0.0.0.0",
port=port,
reload=False,
log_level="info"
)
if __name__ == "__main__":
main()