Authorization Request
-Scope: {html.escape(scope)}
Please try connecting again.
", status_code=400) - - redirect_uri = pending["redirect_uri"] - state = pending["state"] - - if action != "approve": - qs = urlencode({"error": "access_denied", "state": state}) - return RedirectResponse(f"{redirect_uri}?{qs}", status_code=302) - - if not GATEWAY_PASSWORD: - return HTMLResponse("OAUTH_PASSWORD not set.
", status_code=500) - - if password != GATEWAY_PASSWORD: - PENDING_AUTH[internal_state] = pending - client = REGISTERED_CLIENTS.get(pending["client_id"], {}) - client_name = client.get("client_name", "Unknown Client") - return HTMLResponse(render_consent_page(client_name, pending["scope"], internal_state, "Invalid password. Please try again.")) - - code = secrets.token_urlsafe(48) - AUTH_CODES[code] = { - "client_id": pending["client_id"], - "redirect_uri": redirect_uri, - "code_challenge": pending["code_challenge"], - "code_challenge_method": pending["code_challenge_method"], - "scope": pending["scope"], - "expires_at": time.time() + AUTH_CODE_TTL, - "user": "owner", - } - - qs = urlencode({"code": code, "state": state}) - logger.info(f"OAuth: issued auth code for client {pending['client_id']}") - return RedirectResponse(f"{redirect_uri}?{qs}", status_code=302) - - return JSONResponse({"error": "method_not_allowed"}, status_code=405) - - -async def oauth_token(request: Request) -> JSONResponse: - _clean_expired() - - content_type = request.headers.get("content-type", "") - if "application/json" in content_type: - try: - body = await request.json() - except Exception: - body = {} - else: - form = await request.form() - body = dict(form) - - grant_type = body.get("grant_type", "") - - if grant_type == "authorization_code": - code = body.get("code", "") - client_id = body.get("client_id", "") - code_verifier = body.get("code_verifier", "") - - code_info = AUTH_CODES.pop(code, None) - if not code_info: - return JSONResponse({"error": "invalid_grant", "error_description": "Code invalid or expired."}, status_code=400) - - if code_info["expires_at"] < time.time(): - return JSONResponse({"error": "invalid_grant", "error_description": "Code expired."}, status_code=400) - - if code_info["client_id"] != client_id: - return JSONResponse({"error": "invalid_grant", "error_description": "Client mismatch."}, status_code=400) - - if code_info.get("code_challenge"): - if not code_verifier: - return JSONResponse({"error": "invalid_grant", "error_description": "code_verifier required."}, status_code=400) - expected = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).rstrip(b"=").decode() - if expected != code_info["code_challenge"]: - return JSONResponse({"error": "invalid_grant", "error_description": "PKCE verification failed."}, status_code=400) - - access_token = secrets.token_urlsafe(48) - refresh_token = secrets.token_urlsafe(48) - access_hash = _hash(access_token) - refresh_hash = _hash(refresh_token) - - ACCESS_TOKENS[access_hash] = { - "client_id": client_id, - "scope": code_info["scope"], - "expires_at": time.time() + ACCESS_TOKEN_TTL, - "user": code_info["user"], - } - REFRESH_TOKENS[refresh_hash] = { - "client_id": client_id, - "scope": code_info["scope"], - "expires_at": time.time() + REFRESH_TOKEN_TTL, - "user": code_info["user"], - "access_token_hash": access_hash, - } - - logger.info(f"OAuth: issued access token for client {client_id}") - return JSONResponse({ - "access_token": access_token, - "token_type": "Bearer", - "expires_in": ACCESS_TOKEN_TTL, - "refresh_token": refresh_token, - "scope": code_info["scope"], - }) - - elif grant_type == "refresh_token": - refresh_token_val = body.get("refresh_token", "") - client_id = body.get("client_id", "") - - refresh_hash = _hash(refresh_token_val) - refresh_info = REFRESH_TOKENS.get(refresh_hash) - - if not refresh_info: - return JSONResponse({"error": "invalid_grant", "error_description": "Refresh token invalid."}, status_code=400) - - if refresh_info["expires_at"] < time.time(): - del REFRESH_TOKENS[refresh_hash] - return JSONResponse({"error": "invalid_grant", "error_description": "Refresh token expired."}, status_code=400) - - if refresh_info["client_id"] != client_id: - return JSONResponse({"error": "invalid_grant", "error_description": "Client mismatch."}, status_code=400) - - old_access_hash = refresh_info.get("access_token_hash") - if old_access_hash and old_access_hash in ACCESS_TOKENS: - del ACCESS_TOKENS[old_access_hash] - - del REFRESH_TOKENS[refresh_hash] - - new_access_token = secrets.token_urlsafe(48) - new_refresh_token = secrets.token_urlsafe(48) - new_access_hash = _hash(new_access_token) - new_refresh_hash = _hash(new_refresh_token) - - ACCESS_TOKENS[new_access_hash] = { - "client_id": client_id, - "scope": refresh_info["scope"], - "expires_at": time.time() + ACCESS_TOKEN_TTL, - "user": refresh_info["user"], - } - REFRESH_TOKENS[new_refresh_hash] = { - "client_id": client_id, - "scope": refresh_info["scope"], - "expires_at": time.time() + REFRESH_TOKEN_TTL, - "user": refresh_info["user"], - "access_token_hash": new_access_hash, - } - - logger.info(f"OAuth: refreshed token for client {client_id}") - return JSONResponse({ - "access_token": new_access_token, - "token_type": "Bearer", - "expires_in": ACCESS_TOKEN_TTL, - "refresh_token": new_refresh_token, - "scope": refresh_info["scope"], - }) - - return JSONResponse({"error": "unsupported_grant_type"}, status_code=400) - - -# --------------------------------------------------------------------------- -# MCP Endpoint (OAuth-protected) -# --------------------------------------------------------------------------- - -async def handle_mcp(request: Request) -> Response: - if request.method == "HEAD": - return Response(status_code=200, headers={"MCP-Protocol-Version": "2024-11-05"}) - - if request.method == "GET": - token_info = validate_bearer_token(request) - if not token_info: - return Response( - status_code=401, - headers={ - "WWW-Authenticate": f'Bearer resource_metadata="{ISSUER_URL}/.well-known/oauth-protected-resource"', - "Content-Type": "application/json", - }, - media_type="application/json", - content=json.dumps({"error": "unauthorized", "message": "Bearer token required."}), - ) - # Authenticated GET — return empty SSE stream (mcpo uses this for server-sent events) - async def empty_sse(): - yield b": ping\n\n" - return StreamingResponse(empty_sse(), media_type="text/event-stream", - headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}) - - if request.method == "DELETE": - session_id = request.headers.get("Mcp-Session-Id") - if session_id and session_id in GATEWAY_SESSIONS: - del GATEWAY_SESSIONS[session_id] - return Response(status_code=200) - - # POST — require Bearer token - token_info = validate_bearer_token(request) - if not token_info: - return Response( - status_code=401, - headers={ - "WWW-Authenticate": f'Bearer resource_metadata="{ISSUER_URL}/.well-known/oauth-protected-resource"', - "Content-Type": "application/json", - }, - media_type="application/json", - content=json.dumps({"error": "unauthorized", "message": "Valid Bearer token required."}), - ) - - try: - body = await request.json() - except Exception: - return JSONResponse( - {"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None}, - status_code=400, - ) - - method = body.get("method", "") - params = body.get("params", {}) - req_id = body.get("id") - - if method == "initialize": - session_id = str(uuid.uuid4()) - GATEWAY_SESSIONS[session_id] = True - response = JSONResponse({ - "jsonrpc": "2.0", - "id": req_id, - "result": { - "protocolVersion": "2024-11-05", - "capabilities": {"tools": {"listChanged": False}}, - "serverInfo": {"name": "mcp-gateway-proxy", "version": "1.0.0"}, - }, - }) - response.headers["Mcp-Session-Id"] = session_id - return response - - if method == "notifications/initialized": - return Response(status_code=202) - - if method == "tools/list": - tools = list(TOOL_DEFINITIONS.values()) - response = JSONResponse({ - "jsonrpc": "2.0", - "id": req_id, - "result": {"tools": tools}, - }) - session_id = request.headers.get("Mcp-Session-Id") - if session_id: - response.headers["Mcp-Session-Id"] = session_id - return response - - if method == "tools/call": - tool_name = params.get("name", "") - arguments = params.get("arguments", {}) - - backend_name = TOOL_REGISTRY.get(tool_name) - if not backend_name: - return JSONResponse({ - "jsonrpc": "2.0", - "id": req_id, - "error": {"code": -32601, "message": f"Unknown tool: {tool_name}"}, - }) - - try: - result = await forward_tool_call(backend_name, tool_name, arguments, req_id) - - if isinstance(result, dict) and "jsonrpc" in result: - response = JSONResponse(result) - elif isinstance(result, dict) and "result" in result: - response = JSONResponse({ - "jsonrpc": "2.0", - "id": req_id, - "result": result["result"], - }) - elif isinstance(result, dict) and "error" in result: - response = JSONResponse({ - "jsonrpc": "2.0", - "id": req_id, - "error": result["error"], - }) - else: - response = JSONResponse({ - "jsonrpc": "2.0", - "id": req_id, - "result": result, - }) - - session_id = request.headers.get("Mcp-Session-Id") - if session_id: - response.headers["Mcp-Session-Id"] = session_id - return response - - except Exception as e: - logger.error(f"Tool call failed: {tool_name} - {e}") - return JSONResponse({ - "jsonrpc": "2.0", - "id": req_id, - "error": {"code": -32603, "message": str(e)}, - }) - - if method == "ping": - return JSONResponse({"jsonrpc": "2.0", "id": req_id, "result": {}}) - - return JSONResponse({ - "jsonrpc": "2.0", - "id": req_id, - "error": {"code": -32601, "message": f"Method not found: {method}"}, - }) - - -# --------------------------------------------------------------------------- -# Health / Status -# --------------------------------------------------------------------------- - -async def health(request: Request) -> JSONResponse: - return JSONResponse({ - "status": "healthy", - "backends": len(BACKENDS), - "tools": len(TOOL_DEFINITIONS), - "oauth": "enabled", - "active_tokens": len(ACCESS_TOKENS), - "registered_clients": len(REGISTERED_CLIENTS), - }) - - -async def status(request: Request) -> JSONResponse: - token_info = validate_bearer_token(request) - if not token_info: - return JSONResponse({"error": "unauthorized"}, status_code=401) - - return JSONResponse({ - "backends": { - name: { - "url": url, - "tools": len([t for t, b in TOOL_REGISTRY.items() if b == name]), - "session": BACKEND_SESSIONS.get(name), - } - for name, url in BACKENDS.items() - }, - "total_tools": len(TOOL_DEFINITIONS), - "tool_list": sorted(TOOL_DEFINITIONS.keys()), - }) - - -# --------------------------------------------------------------------------- -# Dashboard -# --------------------------------------------------------------------------- - -DISPLAY_NAMES = { - "erpnext": "ERPNext", - "truenas": "TrueNAS", - "homeassistant": "Home Assistant", - "wave": "Wave Finance", - "linkedin": "LinkedIn", -} - -async def probe_backend(name: str, url: str) -> dict: - """Live-probe a backend: initialize it and count its tools in real time.""" - start = time.time() - try: - result = await mcp_request(url, "initialize", { - "protocolVersion": "2024-11-05", - "capabilities": {}, - "clientInfo": {"name": "mcp-gateway-dashboard", "version": "1.0.0"}, - }) - session_id = result.get("session_id") - tools_result = await mcp_request(url, "tools/list", {}, request_id=2, session_id=session_id) - tools_data = tools_result.get("result", {}) - if isinstance(tools_data, dict): - if "result" in tools_data: - tools_data = tools_data["result"] - tools = tools_data.get("tools", []) - else: - tools = [] - elapsed = round((time.time() - start) * 1000) - return { - "status": "healthy", - "toolCount": len(tools), - "responseTime": elapsed, - } - except Exception as e: - elapsed = round((time.time() - start) * 1000) - logger.warning(f"Dashboard probe failed for {name}: {e}") - # Fall back to cached registry values if live probe fails - cached_tools = len([t for t, b in TOOL_REGISTRY.items() if b == name]) - return { - "status": "healthy" if cached_tools > 0 else "unhealthy", - "toolCount": cached_tools, - "responseTime": elapsed, - "note": "cached", - } - - -async def dashboard_status(request: Request) -> JSONResponse: - """Public endpoint — no auth required — live-probes all backends.""" - probes = await asyncio.gather( - *[probe_backend(name, url) for name, url in BACKENDS.items()], - return_exceptions=True - ) - services = [] - for (name, url), probe in zip(BACKENDS.items(), probes): - if isinstance(probe, Exception): - probe = {"status": "unhealthy", "toolCount": 0, "responseTime": None} - services.append({ - "name": DISPLAY_NAMES.get(name, name.capitalize()), - "key": name, - "url": url, - "status": probe["status"], - "toolCount": probe["toolCount"], - "responseTime": probe.get("responseTime"), - "lastCheck": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - }) - return JSONResponse({ - "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), - "services": services, - "summary": { - "total": len(services), - "healthy": len([s for s in services if s["status"] == "healthy"]), - "unhealthy": len([s for s in services if s["status"] == "unhealthy"]), - "totalTools": sum(s["toolCount"] for s in services), - } - }) - - -DASHBOARD_HTML = """ - - - - -Service Status Dashboard
-Total Services
-—
-Total Tools
-—
-Healthy
-—
-