@router.get("/callback")
async def oauth_callback(
request: Request,
code: str | None = Query(None, description="Authorization code"),
state: str | None = Query(None, description="State parameter"),
error: str | None = Query(None, description="OAuth error"),
error_description: str | None = Query(None, description="OAuth error description"),
) -> HTMLResponse:
"""Handle OAuth callback from Claude authentication.
This endpoint receives the authorization code from Claude's OAuth flow
and exchanges it for access tokens.
"""
try:
if error:
error_msg = error_description or error or "OAuth authentication failed"
logger.error(
"OAuth callback error",
error_type="oauth_error",
error_message=error_msg,
oauth_error=error,
oauth_error_description=error_description,
state=state,
operation="oauth_callback",
)
# Update pending flow if state is provided
if state and state in _pending_flows:
_pending_flows[state].update(
{
"completed": True,
"success": False,
"error": error_msg,
}
)
return HTMLResponse(
content=f"""
<html>
<head><title>Login Failed</title></head>
<body>
<h1>Login Failed</h1>
<p>Error: {error_msg}</p>
<p>You can close this window and try again.</p>
</body>
</html>
""",
status_code=400,
)
if not code:
error_msg = "No authorization code received"
logger.error(
"OAuth callback missing authorization code",
error_type="missing_code",
error_message=error_msg,
state=state,
operation="oauth_callback",
)
if state and state in _pending_flows:
_pending_flows[state].update(
{
"completed": True,
"success": False,
"error": error_msg,
}
)
return HTMLResponse(
content=f"""
<html>
<head><title>Login Failed</title></head>
<body>
<h1>Login Failed</h1>
<p>Error: {error_msg}</p>
<p>You can close this window and try again.</p>
</body>
</html>
""",
status_code=400,
)
if not state:
error_msg = "Missing state parameter"
logger.error(
"OAuth callback missing state parameter",
error_type="missing_state",
error_message=error_msg,
operation="oauth_callback",
)
return HTMLResponse(
content=f"""
<html>
<head><title>Login Failed</title></head>
<body>
<h1>Login Failed</h1>
<p>Error: {error_msg}</p>
<p>You can close this window and try again.</p>
</body>
</html>
""",
status_code=400,
)
# Check if this is a valid pending flow
if state not in _pending_flows:
error_msg = "Invalid or expired state parameter"
logger.error(
"OAuth callback with invalid state",
error_type="invalid_state",
error_message="Invalid or expired state parameter",
state=state,
operation="oauth_callback",
)
return HTMLResponse(
content=f"""
<html>
<head><title>Login Failed</title></head>
<body>
<h1>Login Failed</h1>
<p>Error: {error_msg}</p>
<p>You can close this window and try again.</p>
</body>
</html>
""",
status_code=400,
)
# Get flow details
flow = _pending_flows[state]
code_verifier = flow["code_verifier"]
custom_paths = flow["custom_paths"]
# Exchange authorization code for tokens
success = await _exchange_code_for_tokens(code, code_verifier, custom_paths)
# Update flow result
_pending_flows[state].update(
{
"completed": True,
"success": success,
"error": None if success else "Token exchange failed",
}
)
if success:
logger.info(
"OAuth login successful", state=state, operation="oauth_callback"
)
return HTMLResponse(
content="""
<html>
<head><title>Login Successful</title></head>
<body>
<h1>Login Successful!</h1>
<p>You have successfully logged in to Claude.</p>
<p>You can close this window and return to the CLI.</p>
<script>
setTimeout(() => {
window.close();
}, 3000);
</script>
</body>
</html>
""",
status_code=200,
)
else:
error_msg = "Failed to exchange authorization code for tokens"
logger.error(
"OAuth token exchange failed",
error_type="token_exchange_failed",
error_message=error_msg,
state=state,
operation="oauth_callback",
)
return HTMLResponse(
content=f"""
<html>
<head><title>Login Failed</title></head>
<body>
<h1>Login Failed</h1>
<p>Error: {error_msg}</p>
<p>You can close this window and try again.</p>
</body>
</html>
""",
status_code=500,
)
except Exception as e:
logger.error(
"Unexpected error in OAuth callback",
error_type="unexpected_error",
error_message=str(e),
state=state,
operation="oauth_callback",
exc_info=True,
)
if state and state in _pending_flows:
_pending_flows[state].update(
{
"completed": True,
"success": False,
"error": str(e),
}
)
return HTMLResponse(
content=f"""
<html>
<head><title>Login Error</title></head>
<body>
<h1>Login Error</h1>
<p>An unexpected error occurred: {str(e)}</p>
<p>You can close this window and try again.</p>
</body>
</html>
""",
status_code=500,
)