Spaces:
Sleeping
Sleeping
File size: 5,224 Bytes
2415446 9857ebd 2415446 9857ebd 2415446 0a54372 2415446 9857ebd 2415446 0a54372 2415446 0a54372 2415446 | 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 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | """Pure FastAPI application factory."""
from typing import Any
from fastapi import FastAPI, Request
from fastapi.exception_handlers import request_validation_exception_handler
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, Response
from starlette.middleware.base import BaseHTTPMiddleware
from loguru import logger
class ScannerTrapMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
path = request.url.path
if path.startswith("/.env") or path.startswith("/.git") or "wp-" in path:
return Response(content="OK", status_code=200, media_type="text/plain")
return await call_next(request)
from free_claude_code.application.errors import ApplicationError
from free_claude_code.core.anthropic import anthropic_error_payload
from free_claude_code.core.diagnostics import (
redacted_exception_traceback,
safe_exception_message,
)
from free_claude_code.core.openai_responses import openai_error_payload
from free_claude_code.core.trace import (
extract_claude_session_id_from_headers,
trace_event,
)
from free_claude_code.core.version import package_version
from .admin_cache import AdminNoStoreMiddleware, attach_admin_no_store
from .admin_routes import router as admin_router
from .ports import ApiServices
from .request_errors import ordinary_application_error_response
from .request_ids import (
RequestCorrelationMiddleware,
attach_request_id_headers,
get_request_id,
)
from .routes import router
from .validation_log import summarize_request_validation_body
def create_app(services: ApiServices) -> FastAPI:
"""Create the HTTP adapter around explicitly supplied runtime services."""
app = FastAPI(title="Claude Code Proxy", version=package_version())
app.state.services = services
app.add_middleware(ScannerTrapMiddleware)
app.add_middleware(RequestCorrelationMiddleware)
app.add_middleware(AdminNoStoreMiddleware)
app.include_router(admin_router)
app.include_router(router)
@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
"""Log request shape for 422 debugging without content values."""
body: Any
try:
body = await request.json()
except Exception as error:
body = {"_json_error": type(error).__name__}
message_summary, tool_names = summarize_request_validation_body(body)
trace_event(
stage="ingress",
event="server.request.validation_failed",
source="api",
path=request.url.path,
query=dict(request.query_params),
error_locs=[list(error.get("loc", ())) for error in exc.errors()],
error_types=[str(error.get("type", "")) for error in exc.errors()],
message_summary=message_summary,
tool_names=tool_names,
)
return await request_validation_exception_handler(request, exc)
@app.exception_handler(ApplicationError)
async def application_error_handler(request: Request, exc: ApplicationError):
"""Serialize defensive application failures in the selected wire protocol."""
return ordinary_application_error_response(
exc,
wire_api=(
"responses" if request.url.path == "/v1/responses" else "messages"
),
request_id=get_request_id(request),
)
@app.exception_handler(Exception)
async def general_error_handler(request: Request, exc: Exception):
"""Handle general errors and return Anthropic format."""
request_id = get_request_id(request)
claude_sid = extract_claude_session_id_from_headers(request.headers)
settings = services.requests.current_settings()
with logger.contextualize(
http_method=request.method,
http_path=request.url.path,
claude_session_id=claude_sid,
request_id=request_id,
):
if settings.log_api_error_tracebacks:
logger.error("General Error: {}", safe_exception_message(exc))
logger.error(redacted_exception_traceback(exc))
else:
logger.error(
"General Error: path={} method={} exc_type={}",
request.url.path,
request.method,
type(exc).__name__,
)
message = safe_exception_message(exc)
if request.url.path == "/v1/responses":
content = openai_error_payload(message=message, error_type="api_error")
else:
content = anthropic_error_payload(
error_type="api_error",
message=message,
request_id=request_id,
)
response = JSONResponse(status_code=500, content=content)
attach_admin_no_store(response, path=request.url.path)
attach_request_id_headers(
response,
request_id=request_id,
path=request.url.path,
)
return response
return app
|