"""HTTP server for 42crunch MCP Server using FastAPI.
This server exposes the MCP tools via JSON-RPC 2.0 over HTTP.
"""
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, Dict, Any, List
import uvicorn
from .client import FortyTwoCrunchClient
from .config import Config
# Initialize FastAPI app
app = FastAPI(
title="42crunch MCP Server",
description="JSON-RPC 2.0 HTTP server for 42crunch API",
version="1.0.0"
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Configure appropriately for production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Initialize client (will be created on first use)
_client: Optional[FortyTwoCrunchClient] = None
def get_client() -> FortyTwoCrunchClient:
"""Get or create the API client instance."""
global _client
if _client is None:
_client = FortyTwoCrunchClient()
return _client
# JSON-RPC 2.0 Request/Response Models
class JSONRPCRequest(BaseModel):
"""JSON-RPC 2.0 request model."""
jsonrpc: str = "2.0"
id: Optional[Any] = None
method: str
params: Optional[Dict[str, Any]] = None
class JSONRPCError(BaseModel):
"""JSON-RPC 2.0 error model."""
code: int
message: str
data: Optional[Any] = None
class JSONRPCResponse(BaseModel):
"""JSON-RPC 2.0 response model."""
jsonrpc: str = "2.0"
id: Optional[Any] = None
result: Optional[Any] = None
error: Optional[JSONRPCError] = None
# Health check endpoint
@app.get("/health")
async def health_check():
"""Health check endpoint."""
return {"status": "healthy", "service": "42crunch-mcp-server"}
@app.get("/")
async def root():
"""Root endpoint with API information."""
return {
"service": "42crunch MCP Server",
"version": "1.0.0",
"protocol": "JSON-RPC 2.0",
"endpoints": {
"jsonrpc": "/jsonrpc",
"health": "/health",
"tools": "/tools"
}
}
@app.get("/tools")
async def list_tools():
"""List all available tools."""
return {
"tools": [
{
"name": "list_collections",
"description": "List all API collections from 42crunch",
"parameters": {
"page": {"type": "integer", "optional": True, "default": 1},
"per_page": {"type": "integer", "optional": True, "default": 10},
"order": {"type": "string", "optional": True, "default": "default"},
"sort": {"type": "string", "optional": True, "default": "default"}
}
},
{
"name": "get_collection_apis",
"description": "Get all APIs within a collection",
"parameters": {
"collection_id": {"type": "string", "required": True},
"with_tags": {"type": "boolean", "optional": True, "default": True}
}
},
{
"name": "get_api_details",
"description": "Get detailed information about a specific API",
"parameters": {
"api_id": {"type": "string", "required": True},
"branch": {"type": "string", "optional": True, "default": "main"},
"include_definition": {"type": "boolean", "optional": True, "default": True},
"include_assessment": {"type": "boolean", "optional": True, "default": True},
"include_scan": {"type": "boolean", "optional": True, "default": True}
}
}
]
}
@app.post("/jsonrpc")
async def jsonrpc_endpoint(request: JSONRPCRequest):
"""Main JSON-RPC 2.0 endpoint."""
try:
# Validate JSON-RPC version
if request.jsonrpc != "2.0":
return JSONRPCResponse(
id=request.id,
error=JSONRPCError(
code=-32600,
message="Invalid Request",
data="jsonrpc version must be '2.0'"
)
).dict(exclude_none=True)
# Route to appropriate handler
if request.method == "tools/list":
return handle_tools_list(request.id)
elif request.method == "tools/call":
return handle_tools_call(request.id, request.params or {})
elif request.method == "initialize":
return handle_initialize(request.id, request.params or {})
elif request.method == "ping":
return handle_ping(request.id)
else:
return JSONRPCResponse(
id=request.id,
error=JSONRPCError(
code=-32601,
message="Method not found",
data=f"Unknown method: {request.method}"
)
).dict(exclude_none=True)
except Exception as e:
return JSONRPCResponse(
id=request.id if hasattr(request, 'id') else None,
error=JSONRPCError(
code=-32603,
message="Internal error",
data=str(e)
)
).dict(exclude_none=True)
def handle_initialize(request_id: Any, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle initialize request."""
return JSONRPCResponse(
id=request_id,
result={
"protocolVersion": "2024-11-05",
"capabilities": {},
"serverInfo": {
"name": "42crunch-mcp-server",
"version": "1.0.0"
}
}
).dict(exclude_none=True)
def handle_ping(request_id: Any) -> Dict[str, Any]:
"""Handle ping request."""
return JSONRPCResponse(
id=request_id,
result={"status": "pong"}
).dict(exclude_none=True)
def handle_tools_list(request_id: Any) -> Dict[str, Any]:
"""Handle tools/list request."""
tools = [
{
"name": "list_collections",
"description": "List all API collections from 42crunch",
"inputSchema": {
"type": "object",
"properties": {
"page": {"type": "integer", "default": 1},
"per_page": {"type": "integer", "default": 10},
"order": {"type": "string", "default": "default"},
"sort": {"type": "string", "default": "default"}
}
}
},
{
"name": "get_collection_apis",
"description": "Get all APIs within a collection",
"inputSchema": {
"type": "object",
"required": ["collection_id"],
"properties": {
"collection_id": {"type": "string"},
"with_tags": {"type": "boolean", "default": True}
}
}
},
{
"name": "get_api_details",
"description": "Get detailed information about a specific API",
"inputSchema": {
"type": "object",
"required": ["api_id"],
"properties": {
"api_id": {"type": "string"},
"branch": {"type": "string", "default": "main"},
"include_definition": {"type": "boolean", "default": True},
"include_assessment": {"type": "boolean", "default": True},
"include_scan": {"type": "boolean", "default": True}
}
}
}
]
return JSONRPCResponse(
id=request_id,
result={"tools": tools}
).dict(exclude_none=True)
def handle_tools_call(request_id: Any, params: Dict[str, Any]) -> Dict[str, Any]:
"""Handle tools/call request."""
tool_name = params.get("name")
arguments = params.get("arguments", {})
if not tool_name:
return JSONRPCResponse(
id=request_id,
error=JSONRPCError(
code=-32602,
message="Invalid params",
data="Missing required parameter: name"
)
).dict(exclude_none=True)
try:
client = get_client()
if tool_name == "list_collections":
result = client.list_collections(
page=arguments.get("page", 1),
per_page=arguments.get("per_page", 10),
order=arguments.get("order", "default"),
sort=arguments.get("sort", "default"),
)
return JSONRPCResponse(
id=request_id,
result={
"success": True,
"data": result
}
).dict(exclude_none=True)
elif tool_name == "get_collection_apis":
collection_id = arguments.get("collection_id")
if not collection_id:
return JSONRPCResponse(
id=request_id,
error=JSONRPCError(
code=-32602,
message="Invalid params",
data="Missing required parameter: collection_id"
)
).dict(exclude_none=True)
result = client.get_collection_apis(
collection_id=collection_id,
with_tags=arguments.get("with_tags", True),
)
return JSONRPCResponse(
id=request_id,
result={
"success": True,
"data": result
}
).dict(exclude_none=True)
elif tool_name == "get_api_details":
api_id = arguments.get("api_id")
if not api_id:
return JSONRPCResponse(
id=request_id,
error=JSONRPCError(
code=-32602,
message="Invalid params",
data="Missing required parameter: api_id"
)
).dict(exclude_none=True)
result = client.get_api_details(
api_id=api_id,
branch=arguments.get("branch", "main"),
read_tags=True,
read_openapi_definition=arguments.get("include_definition", True),
read_assessment=arguments.get("include_assessment", True),
read_scan=arguments.get("include_scan", True),
)
return JSONRPCResponse(
id=request_id,
result={
"success": True,
"data": result
}
).dict(exclude_none=True)
else:
return JSONRPCResponse(
id=request_id,
error=JSONRPCError(
code=-32601,
message="Method not found",
data=f"Unknown tool: {tool_name}"
)
).dict(exclude_none=True)
except Exception as e:
return JSONRPCResponse(
id=request_id,
error=JSONRPCError(
code=-32603,
message="Internal error",
data={
"error": str(e),
"error_type": type(e).__name__
}
)
).dict(exclude_none=True)
def run_server(host: str = "0.0.0.0", port: int = 8000, reload: bool = False):
"""Run the HTTP server.
Args:
host: Host to bind to
port: Port to bind to
reload: Enable auto-reload for development
"""
uvicorn.run(
"src.http_server:app",
host=host,
port=port,
reload=reload,
log_level="info"
)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="42crunch MCP HTTP Server")
parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
parser.add_argument("--port", type=int, default=8000, help="Port to bind to")
parser.add_argument("--reload", action="store_true", help="Enable auto-reload")
args = parser.parse_args()
run_server(host=args.host, port=args.port, reload=args.reload)
