""" BeamBridge Enterprise Platform — FastAPI Backend & Live Anthropic MCP v2 JSON-RPC 2.0 Endpoint. Serves the Crextio-designed executive console on 0.0.0.0:7860, backed 100% by real cloned GitHub repositories, official Beam-1.0 BPE tokenization, real OpenRouter `gen-...` telemetry, and autonomously synthesized standalone `MCPServer` modules. """ from __future__ import annotations import json from pathlib import Path from typing import Any from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from backend.ast_analyzer import analyze_repository, ingest_live_github_repository from backend.beam_client import synthesize_with_beam from backend.catalog import REPO_CATALOG from backend.finops_roi import calculate_enterprise_roi from backend.parity_verifier import ( ORACLE_DISPATCH, get_empirical_ledger, invoke_synthesized_mcp_tool, list_all_synthesized_mcp_tools, run_all_verifications, run_repo_verification, ) from backend.real_oracles import load_real_repo_files BASE_DIR = Path(__file__).resolve().parent.parent FRONTEND_DIR = BASE_DIR / "frontend" app = FastAPI( title="BeamBridge — Enterprise Legacy-to-MCP Modernization & Verification", version="2.0.0", description="Powered by Reflection AI Beam-1.0-501B-A23B (1M Context · 23B Active MoE · Official 151,681 BPE Tokenizer)", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.middleware("http") async def disable_static_asset_cache(request: Request, call_next): response = await call_next(request) if request.url.path == "/" or request.url.path.startswith("/assets/"): response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0" response.headers["Pragma"] = "no-cache" response.headers["Expires"] = "0" return response class FinOpsQuery(BaseModel): legacy_kloc: float = Field(default=250.0, ge=10.0, le=10000.0) microservices_count: int = Field(default=18, ge=1, le=500) engineer_fully_loaded_annual_usd: float = Field(default=185000.0, ge=50000.0, le=600000.0) daily_agent_verification_runs: int = Field(default=120, ge=5, le=10000) class BeamSynthesisRequest(BaseModel): repo_id: str = Field(default="py_moneyed") provider: str = Field(default="openrouter") api_key: str | None = Field(default=None) custom_base_url: str | None = Field(default=None) reasoning_effort: str = Field(default="medium") preserve_thinking: bool = Field(default=True) user_directive: str | None = Field(default=None) class GitHubIngestRequest(BaseModel): github_url: str = Field(..., examples=["https://github.com/limist/py-moneyed"]) max_files: int = Field(default=120, ge=5, le=400) class MCPJsonRpcRequest(BaseModel): jsonrpc: str = "2.0" id: int | str | None = 1 method: str params: dict[str, Any] = Field(default_factory=dict) @app.get("/api/health") def health() -> dict[str, Any]: ledger = get_empirical_ledger() return { "status": "ok", "platform": "BeamBridge v2.0 (Zero-Simulation Real Empirical Architecture)", "model": "reflection/Beam-1.0-501B-A23B", "architecture": "501B Total / 23B Active MoE · 1,048,576 Context · Official 151,681 BPE Vocab", "license": "Apache-2.0", "ledger_generated_at": ledger.get("generated_at_utc"), } @app.get("/api/bootstrap") def bootstrap_dashboard() -> dict[str, Any]: verification_summary = run_all_verifications() finops_default = calculate_enterprise_roi() ledger = get_empirical_ledger() repos_payload: list[dict[str, Any]] = [] total_raw_tokens = 0 total_packed_tokens = 0 total_loc = 0 total_files = 0 for repo_id, meta in REPO_CATALOG.items(): ast_info = analyze_repository(repo_id) ver_info = verification_summary["repositories"][repo_id] total_raw_tokens += ast_info["raw_disk_beam_tokens"] total_packed_tokens += ast_info["estimated_beam_tokens"] total_loc += ast_info["total_loc"] total_files += ast_info["raw_disk_file_count"] repos_payload.append({ **meta, "ast": ast_info, "verification": { "total_cases": ver_info["total_cases"], "beam_1m_passed": ver_info["beam_1m_passed"], "beam_1m_parity_pct": ver_info["beam_1m_parity_pct"], "chunked_rag_passed": ver_info["chunked_rag_passed"], "chunked_rag_parity_pct": ver_info["chunked_rag_parity_pct"], "parity_lift_pct_points": ver_info["parity_lift_pct_points"], "certificate_sha256": ver_info["certificate_sha256"], "avg_mcp_latency_us": ver_info["avg_mcp_latency_us"], "synthesis_audit": ver_info["synthesis_audit"], }, }) return { "model_architecture": { "name": "Beam-1.0-501B-A23B", "total_params": "501B", "active_params": "23B", "context_window": 1_048_576, "tokenizer_vocab": 151_681, "layers": "52 Layers (3 Dense + 49 MoE · 256 Experts, Top-8 Routed)", "attention": "Hybrid 43 Sliding-Window RoPE + 9 Global No-PE Layers · Scalable Softmax", "license": "Apache 2.0", }, "portfolio_totals": { "repositories_count": len(repos_payload), "total_files": total_files, "total_loc": total_loc, "total_raw_beam_tokens": total_raw_tokens, "total_beam_tokens": total_packed_tokens, "mcp_tools_exposed": len(list_all_synthesized_mcp_tools()), "total_verification_assertions": verification_summary["total_cases"], "beam_1m_parity_pct": verification_summary["beam_1m_parity_pct"], "chunked_rag_parity_pct": verification_summary["chunked_rag_parity_pct"], "master_certificate_sha256": verification_summary["master_certificate_sha256"], }, "repositories": repos_payload, "head_to_head_benchmark": ledger["head_to_head_benchmark"], "finops": finops_default, "mcp_tools": list_all_synthesized_mcp_tools(), } @app.get("/api/repos/{repo_id}") def get_repo_detail(repo_id: str) -> dict[str, Any]: if repo_id not in REPO_CATALOG: raise HTTPException(status_code=404, detail=f"Unknown repository `{repo_id}`") packed_files, _ = load_real_repo_files(repo_id) mcp_file = BASE_DIR / "modernized_mcp" / f"{repo_id}_mcp.py" modernized_source = mcp_file.read_text(encoding="utf-8") if mcp_file.exists() else "" ledger = get_empirical_ledger() wr = ledger["repositories"][repo_id]["whole_repo_1m"] reasoning_log = "\n\n---\n\n".join( f"[Turn {t['turn']} · {t['model']} · {t['request_id']} · {t['passed']}/{t['total']} ({t['pass_rate_pct']}%)]\n{t['reasoning_excerpt']}" for t in wr["turns"] if t.get("reasoning_excerpt") ) return { "catalog": REPO_CATALOG[repo_id], "ast": analyze_repository(repo_id), "legacy_files": packed_files, "modernized_mcp_code": modernized_source, "recorded_reasoning_trace": reasoning_log, "verification": run_repo_verification(repo_id), } @app.post("/api/verify/{repo_id}") def trigger_repo_verification(repo_id: str) -> dict[str, Any]: if repo_id not in REPO_CATALOG: raise HTTPException(status_code=404, detail=f"Unknown repository `{repo_id}`") return run_repo_verification(repo_id) @app.post("/api/verify-all") def trigger_all_verifications() -> dict[str, Any]: return run_all_verifications() @app.post("/api/finops") def compute_finops(query: FinOpsQuery) -> dict[str, Any]: return calculate_enterprise_roi( legacy_kloc=query.legacy_kloc, microservices_count=query.microservices_count, engineer_fully_loaded_annual_usd=query.engineer_fully_loaded_annual_usd, daily_agent_verification_runs=query.daily_agent_verification_runs, ) @app.post("/api/beam/synthesize") async def live_beam_synthesis(req: BeamSynthesisRequest) -> dict[str, Any]: if req.repo_id not in REPO_CATALOG: raise HTTPException(status_code=404, detail=f"Unknown repository `{req.repo_id}`") return await synthesize_with_beam( repo_id=req.repo_id, api_key=req.api_key, provider=req.provider, custom_base_url=req.custom_base_url, reasoning_effort=req.reasoning_effort, preserve_thinking=req.preserve_thinking, user_directive=req.user_directive, ) @app.post("/api/github/ingest") def live_github_ingest(req: GitHubIngestRequest) -> dict[str, Any]: try: return ingest_live_github_repository(req.github_url, max_files=req.max_files) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc @app.get("/api/mcp/tools") def get_mcp_tools() -> dict[str, Any]: tools = list_all_synthesized_mcp_tools() return {"protocolVersion": "2025-06-18", "sdk": "mcp v2 (MCPServer)", "tools": tools} @app.post("/mcp") def handle_mcp_jsonrpc(rpc: MCPJsonRpcRequest) -> JSONResponse: """ Standards-compliant Model Context Protocol (MCP) JSON-RPC 2.0 HTTP endpoint. Invokes the real standalone synthesized MCPServer tools AND cross-checks live against the untouched cloned GitHub repository oracle! """ method = rpc.method if method == "initialize": return JSONResponse({ "jsonrpc": "2.0", "id": rpc.id, "result": { "protocolVersion": "2025-06-18", "capabilities": {"tools": {"listChanged": False}}, "serverInfo": { "name": "beambridge-verified-mcp-gateway", "version": "2.0.0", }, }, }) if method == "tools/list": return JSONResponse({ "jsonrpc": "2.0", "id": rpc.id, "result": {"tools": list_all_synthesized_mcp_tools()}, }) if method == "tools/call": tool_name = rpc.params.get("name", "") arguments = rpc.params.get("arguments") or {} try: repo_key, result = invoke_synthesized_mcp_tool(tool_name, arguments) oracle_fn = ORACLE_DISPATCH.get(tool_name) oracle_match = None if oracle_fn is not None: try: oracle_out = oracle_fn(**arguments) oracle_match = (oracle_out == result) except Exception: oracle_match = None envelope = { "mcp_tool": tool_name, "source_github_repo": REPO_CATALOG[repo_key]["github_url"], "git_sha": REPO_CATALOG[repo_key]["git_sha"], "live_oracle_verified": oracle_match, "output": result, } return JSONResponse({ "jsonrpc": "2.0", "id": rpc.id, "result": { "content": [{"type": "text", "text": json.dumps(envelope, indent=2)}], "structuredContent": envelope, "isError": False, }, }) except KeyError: return JSONResponse({ "jsonrpc": "2.0", "id": rpc.id, "error": {"code": -32601, "message": f"Unknown MCP tool `{tool_name}`"}, }) except Exception as exc: return JSONResponse({ "jsonrpc": "2.0", "id": rpc.id, "error": {"code": -32602, "message": f"Tool execution error: {exc}"}, }) return JSONResponse({ "jsonrpc": "2.0", "id": rpc.id, "error": {"code": -32601, "message": f"Unsupported JSON-RPC method `{method}`"}, }) class DifferentialInvokeRequest(BaseModel): repo_id: str | None = None tool_name: str arguments: dict[str, Any] = Field(default_factory=dict) @app.post("/api/mcp/invoke") def differential_invoke(req: DifferentialInvokeRequest) -> dict[str, Any]: import time as _time oracle_fn = ORACLE_DISPATCH.get(req.tool_name) if oracle_fn is None: raise HTTPException(status_code=404, detail=f"Unknown tool `{req.tool_name}`") try: t0 = _time.perf_counter_ns() legacy_out = oracle_fn(**req.arguments) legacy_us = round((_time.perf_counter_ns() - t0) / 1000.0, 2) t1 = _time.perf_counter_ns() repo_key, modern_out = invoke_synthesized_mcp_tool(req.tool_name, req.arguments) modern_us = round((_time.perf_counter_ns() - t1) / 1000.0, 2) except Exception as exc: raise HTTPException(status_code=400, detail=f"Execution error: {exc}") from exc parity_ok = (legacy_out == modern_out) rpc_envelope = { "jsonrpc": "2.0", "id": 1, "result": { "content": [{"type": "text", "text": json.dumps(modern_out)}], "structuredContent": { "mcp_tool": req.tool_name, "source_github_repo": REPO_CATALOG[repo_key]["github_url"], "git_sha": REPO_CATALOG[repo_key]["git_sha"], "live_oracle_verified": parity_ok, "output": modern_out, }, "isError": False, }, } return { "repo_id": repo_key, "tool_name": req.tool_name, "legacy_output": legacy_out, "modern_output": modern_out, "legacy_latency_us": legacy_us, "modern_latency_us": modern_us, "parity_verified": parity_ok, "mcp_jsonrpc_response": rpc_envelope, } @app.get("/api/dataset/export") def export_dataset_jsonl() -> FileResponse: ds_file = BASE_DIR / "dataset" / "legacy2mcp_bench.jsonl" if not ds_file.exists(): raise HTTPException(status_code=404, detail="Dataset JSONL not found") return FileResponse(str(ds_file), media_type="application/jsonl", filename="legacy2mcp_bench.jsonl") app.mount("/assets", StaticFiles(directory=str(FRONTEND_DIR)), name="assets") @app.get("/") def serve_spa_index() -> FileResponse: return FileResponse(str(FRONTEND_DIR / "index.html"))