# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import RedirectResponse, JSONResponse, HTMLResponse
from app.component.environment import env
from app.component.oauth_adapter import OauthCallbackPayload, get_oauth_adapter
from typing import Optional
import logging
logger = logging.getLogger("server_oauth_controller")
router = APIRouter(prefix="/oauth", tags=["Oauth Servers"])
@router.get("/{app}/login", name="OAuth Login Redirect")
def oauth_login(app: str, request: Request, state: Optional[str] = None):
"""Redirect user to OAuth provider's authorization endpoint."""
try:
callback_url = str(request.url_for("OAuth Callback", app=app))
if callback_url.startswith("http://"):
callback_url = "https://" + callback_url[len("http://") :]
adapter = get_oauth_adapter(app, callback_url)
url = adapter.get_authorize_url(state)
if not url:
logger.error("Failed to generate authorization URL", extra={"provider": app, "callback_url": callback_url})
raise HTTPException(status_code=400, detail="Failed to generate authorization URL")
logger.info("OAuth login initiated", extra={"provider": app})
return RedirectResponse(str(url))
except HTTPException:
raise
except Exception as e:
logger.error("OAuth login failed", extra={"provider": app, "error": str(e)}, exc_info=True)
raise HTTPException(status_code=400, detail="OAuth login failed")
@router.get("/{app}/callback", name="OAuth Callback")
def oauth_callback(app: str, request: Request, code: Optional[str] = None, state: Optional[str] = None):
"""Handle OAuth provider callback and redirect to client app."""
if not code:
logger.warning("OAuth callback missing code", extra={"provider": app})
raise HTTPException(status_code=400, detail="Missing code parameter")
logger.info("OAuth callback received", extra={"provider": app, "has_state": state is not None})
redirect_url = f"eigent://callback/oauth?provider={app}&code={code}&state={state}"
html_content = f"""
OAuth Callback