Authentication
Secure your AG2 backend with user authentication on /chat
Overview#
CopilotKit supports user authentication for AG2 backends in two deployment modes:
- LangGraph Platform equivalent: managed runtime forwarding to your AG2
/chatendpoint - Self-hosted runtime: your own CopilotKit runtime forwarding to your AG2
/chatendpoint
Both approaches let your AG2 backend access authenticated user context and enforce authorization.
This pattern enables your backend to:
- Validate user tokens before dispatching the agent
- Attach authenticated user context to agent state/tools
- Enforce authorization decisions server-side
CopilotKit consumes AG-UI protocol events streamed by AG2 over /chat. See the AG2 AG-UI integration docs.
How It Works#
sequenceDiagram
participant Frontend
participant CopilotKit
participant AG2Backend
participant Agent
Frontend->>CopilotKit: authorization: "user-token"
CopilotKit->>AG2Backend: Forward auth token header
AG2Backend->>AG2Backend: Validate token
AG2Backend->>Agent: Dispatch with authenticated context
Agent->>Agent: Access authenticated user contextFrontend Setup#
Pass your authentication token via the properties prop:
<CopilotKit
runtimeUrl="/api/copilotkit"
properties={{
authorization: userToken, // forwarded to AG2 /chat
}}
>
<YourApp />
</CopilotKit>Note: The authorization property is forwarded to your AG2 /chat endpoint as a request header.
LangGraph Platform Deployment#
For managed deployments, protect your AG2 /chat endpoint with token-header validation.
Setup Authentication Handler#
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse
from autogen import ConversableAgent, LLMConfig
from autogen.ag_ui import AGUIStream, RunAgentInput
agent = ConversableAgent(
name="assistant",
system_message="You are a helpful assistant.",
llm_config=LLMConfig({"model": "gpt-5.4-mini"}),
human_input_mode="NEVER",
)
stream = AGUIStream(agent)
app = FastAPI()
def validate_your_token(token: str) -> dict:
# Replace this with your own validation logic.
if token != "valid-token":
raise HTTPException(status_code=401, detail="Unauthorized")
return {"user_id": "user_123", "role": "member"}
@app.post("/chat")
async def run_agent(
message: RunAgentInput,
accept: str | None = Header(None),
authorization: str | None = Header(None),
):
if not authorization:
raise HTTPException(status_code=401, detail="Missing authorization header")
token = authorization.replace("Bearer ", "")
user_info = validate_your_token(token)
# Use user_info to scope tools, state, and data access before dispatch.
return StreamingResponse(
stream.dispatch(message, accept=accept),
media_type=accept or "text/event-stream",
)Access User in Agent#
Use validated user identity to scope tool calls and data access:
from typing import Annotated
from autogen import ContextVariables
@agent.register_for_llm(description="Return account data for the authenticated user.")
def get_account_data(
context: ContextVariables,
account_id: Annotated[str, "The target account id"],
) -> dict:
user = context.get("auth_user")
if not user:
return {"error": "unauthorized"}
# Example check: ensure user can access this account
if account_id not in user.get("allowed_accounts", []):
return {"error": "forbidden"}
return {"account_id": account_id, "owner": user["user_id"]}Self-hosted Deployment#
For self-hosted deployments, use the same /chat header-validation pattern in your own FastAPI service.
Setup Dynamic Agent Configuration#
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse
from autogen import ConversableAgent, LLMConfig
from autogen.ag_ui import AGUIStream, RunAgentInput
agent = ConversableAgent(
name="assistant",
system_message="You are a helpful assistant.",
llm_config=LLMConfig({"model": "gpt-5.4-mini"}),
human_input_mode="NEVER",
)
stream = AGUIStream(agent)
app = FastAPI()
@app.post("/chat")
async def run_agent(
message: RunAgentInput,
accept: str | None = Header(None),
authorization: str | None = Header(None),
):
if not authorization:
raise HTTPException(status_code=401, detail="Unauthorized")
# Validate token here, then dispatch
return StreamingResponse(
stream.dispatch(message, accept=accept),
media_type=accept or "text/event-stream",
)Access User in Agent#
After token validation, persist user identity in request-scoped context and enforce access checks in AG2 tools/state reads.
Universal Authentication Pattern#
For backends that run in both managed and self-hosted modes, use this pattern:
def extract_user_from_auth_header(authorization: str | None) -> dict | None:
if not authorization:
return None
token = authorization.replace("Bearer ", "")
return validate_your_token(token)Then:
- Read
authorizationon/chat - Validate token before
stream.dispatch(...) - Attach user context for tool/state authorization
- Deny unauthorized or out-of-scope access
Security Notes#
LangGraph Platform#
- Token Validation: Validate tokens on your AG2
/chatendpoint - User Scoping: Scope data access by authenticated user identity
Self-hosted#
- Manual Validation: Implement and maintain your own validation logic
- Header Forwarding: Ensure your runtime forwards
authorizationto AG2
General Best Practices#
- Permission Checks: Enforce role-based checks in AG2 tools
- Transport Security: Serve
/chatover HTTPS - Least Privilege: Return only data needed for the current user/task
Troubleshooting#
Common Issues#
Token not reaching backend:
- Ensure you're passing
authorizationinproperties - Confirm your runtime forwards headers to AG2
/chat
Invalid token format:
- Handle both raw tokens and
Bearer <token>formats consistently
Unexpected anonymous access:
- Verify
authorizationchecks happen before callingstream.dispatch(...)
