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
+—
+