Spaces:
Runtime error
Runtime error
File size: 2,610 Bytes
cca012f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | import json
import os
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
from pydantic import BaseModel
from typing import Dict, Any, Optional
from tech_radar.mcp.tools import (
get_db_and_vector_store,
tool_search_tech_jobs,
tool_analyze_skill_gap,
tool_generate_resume_patch,
tool_get_market_insights,
tool_generate_interview_prep_kit
)
app = FastAPI(title="TechRadar MCP Web Server")
class McpRequest(BaseModel):
tool: str
args: Dict[str, Any]
@app.get("/", response_class=HTMLResponse)
def get_web_app():
html_path = os.path.join(os.path.dirname(__file__), "web_app.html")
if not os.path.exists(html_path):
raise HTTPException(status_code=404, detail="web_app.html not found")
with open(html_path, "r", encoding="utf-8") as f:
return f.read()
@app.get("/api/jobs")
def get_jobs(domain: Optional[str] = None, city: Optional[str] = None):
db, _, _, _ = get_db_and_vector_store()
jobs = [j.dict() for j in db.search_jobs(domain=domain, city=city, limit=200)]
return {"count": len(jobs), "jobs": jobs}
@app.get("/api/insights")
def get_insights(city: str = "All", domain: str = "All"):
_, _, _, analyst = get_db_and_vector_store()
report = analyst.generate_market_report(city=city, domain=domain)
return report.dict()
@app.post("/api/mcp/execute")
def execute_mcp_tool(req: McpRequest):
tool_name = req.tool
args = req.args
if tool_name == "search_tech_jobs":
res = tool_search_tech_jobs(
domain=args.get("domain", "All"),
city=args.get("city", "All"),
query=args.get("query")
)
elif tool_name == "analyze_skill_gap":
res = tool_analyze_skill_gap(
resume_text=args.get("resume_text", ""),
target_job_id=args.get("target_job_id", "")
)
elif tool_name == "generate_tailored_resume_patch":
res = tool_generate_resume_patch(
resume_text=args.get("resume_text", ""),
target_job_id=args.get("target_job_id", "")
)
elif tool_name == "get_market_insights":
res = tool_get_market_insights(
domain=args.get("domain", "All"),
city=args.get("city", "All")
)
elif tool_name == "generate_interview_prep_kit":
res = tool_generate_interview_prep_kit(
target_job_id=args.get("target_job_id", "")
)
else:
raise HTTPException(status_code=400, detail=f"Unknown tool: {tool_name}")
return {"status": "success", "tool": tool_name, "result": res}
|