showcase/shell-docs/src/content/docs/auth.mdx
You have a chat surface or a hook driving an agent and you want every agent run to know who the request came from. By the end of this guide, your frontend will forward a token, the runtime will pass it through, and your agent code will read the resulting user info on every turn.
If you don't need any of those, skip auth entirely. The agent runs anonymously and the frontend never has to care about tokens.
<WhenFrameworkHas flag="auth_pattern" equals="runtime-onrequest">Pass your token via the headers prop on <CopilotKit>. CopilotKit forwards every request with that header attached.
import { CopilotKit } from "@copilotkit/react-core/v2";
<CopilotKit
runtimeUrl="/api/copilotkit"
headers={{
Authorization: `Bearer ${userToken}`,
}}
>
<YourApp />
</CopilotKit>
Wire authentication into the V2 runtime via the onRequest hook. The hook runs before any agent code and operates on the raw Request, so it's the right place to read the Authorization header, run your verifier, and either let the request through or short-circuit with a 401:
import type { NextRequest } from "next/server";
import {
CopilotRuntime,
createCopilotRuntimeHandler,
} from "@copilotkit/runtime/v2";
const runtime = new CopilotRuntime({ agents: { default: myAgent } });
const handler = createCopilotRuntimeHandler({
runtime,
basePath: "/api/copilotkit",
hooks: {
onRequest: ({ request }) => {
const authHeader = request.headers.get("authorization");
if (!authHeader?.startsWith("Bearer ")) {
throw new Response(
JSON.stringify({ error: "unauthorized" }),
{ status: 401, headers: { "content-type": "application/json" } },
);
}
const token = authHeader.slice("Bearer ".length);
const user = verifyJwt(token); // your validation
// attach user to request-scoped context here
},
},
});
export const POST = (req: NextRequest) => handler(req);
export const GET = (req: NextRequest) => handler(req);
</WhenFrameworkHas> <WhenFrameworkHas flag="auth_pattern" equals="langgraph">The V1 Next.js adapter (
copilotRuntimeNextJSAppRouterEndpoint) does not forward thehooksoption. UsecreateCopilotRuntimeHandlerfrom@copilotkit/runtime/v2directly when you need theonRequestgate.
Pass your token via the headers prop. CopilotKit attaches it to every runtime request, and the runtime forwards the Authorization header on to your agent — whether that's a LangGraph deployment or a self-hosted AG-UI endpoint.
import { CopilotKit } from "@copilotkit/react-core/v2";
<CopilotKit
runtimeUrl="/api/copilotkit"
headers={{
Authorization: `Bearer ${userToken}`,
}}
>
<YourApp />
</CopilotKit>
LangGraph supports two deployment modes. The frontend code above is the same in both, but the backend wiring differs in where the resolved user identity lands. Pick the tab that matches where your agent runs.
<Tabs items={['LangGraph Platform', 'Self-hosted (FastAPI)']}> <Tab value="LangGraph Platform">
On LangGraph Platform (and on langgraph dev), authentication is a managed service. You declare an @auth.authenticate handler, and the server runs it on every request before the graph starts. The forwarded Authorization header arrives as the handler's authorization argument, and the handler's return value becomes available to every node in the run.
from langgraph_sdk import Auth
auth = Auth()
@auth.authenticate
async def authenticate(authorization: str | None):
if not authorization or not authorization.startswith("Bearer "):
raise Auth.exceptions.HTTPException(status_code=401, detail="Unauthorized")
token = authorization.replace("Bearer ", "")
user_info = validate_your_token(token) # your validation logic
return {
"identity": user_info["user_id"],
"role": user_info.get("role"),
"permissions": user_info.get("permissions", []),
}
The return value of the handler shows up in every node's config["configurable"]["langgraph_auth_user"]. From there, scoping tool access or filtering data is straightforward:
from langchain_core.runnables import RunnableConfig
async def my_agent_node(state: AgentState, config: RunnableConfig):
user_info = config["configurable"]["langgraph_auth_user"]
user_id = user_info["identity"]
user_role = user_info.get("role")
# agent logic with user context
return state
For full handler details, see the LangGraph Platform Authentication documentation.
</Tab> <Tab value="Self-hosted (FastAPI)">When you self-host the agent behind FastAPI, there's no managed auth handler to plug into — validation is your job, and the natural place for it is the endpoint that serves the AG-UI stream. add_langgraph_fastapi_endpoint mounts that endpoint for you, but it takes one pre-built agent and gives you no per-request hook, so replace it with the equivalent route of your own: a FastAPI dependency verifies the Authorization header the runtime forwarded, and the resolved user is baked into a per-request agent's config.
from typing import Optional
from ag_ui.core.types import RunAgentInput
from ag_ui.encoder import EventEncoder
from copilotkit import LangGraphAGUIAgent
from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import StreamingResponse
from src.agent import graph
app = FastAPI()
def current_user(authorization: Optional[str] = Header(default=None)) -> dict:
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing bearer token")
return validate_your_token(authorization.removeprefix("Bearer ").strip()) # your validation
@app.post("/")
async def run_agent(
input_data: RunAgentInput,
request: Request,
user: dict = Depends(current_user),
):
encoder = EventEncoder(accept=request.headers.get("accept"))
# One agent per request: the verified identity rides on this run only, and
# each request gets its own isolated streaming state.
agent = LangGraphAGUIAgent(
name="sample_agent",
graph=graph,
config={"configurable": {"auth_user": user}},
)
async def event_generator():
async for event in agent.run(input_data):
yield encoder.encode(event)
return StreamingResponse(event_generator(), media_type=encoder.get_content_type())
Unauthenticated requests never reach the graph — they get a 401 from the dependency. Authenticated ones arrive with an already-verified user on the config, so nodes read identity instead of re-validating a raw token:
from langchain_core.runnables import RunnableConfig
async def my_agent_node(state: AgentState, config: RunnableConfig):
user = config["configurable"]["auth_user"]
user_id = user["user_id"]
user_role = user.get("role")
# agent logic with user context
return state
Pass your token via the properties prop. CopilotKit forwards it to AG2's /chat endpoint as a request header.
import { CopilotKit } from "@copilotkit/react-core/v2";
<CopilotKit
runtimeUrl="/api/copilotkit"
properties={{
authorization: userToken,
}}
>
<YourApp />
</CopilotKit>
The backend has two responsibilities: validate the token before the agent dispatches, and thread the resolved user identity into AG2's ContextVariables so tools can read it later.
Start by validating the token on AG2's /chat endpoint. The Authorization header arrives as a normal FastAPI Header(...) parameter:
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"}),
)
stream = AGUIStream(agent)
app = FastAPI()
def validate_your_token(token: str) -> dict:
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",
)
Once the token is validated, AG2's tools can read the user identity straight out of ContextVariables. This is how you make individual tool calls aware of who's asking, without having to thread the user object manually through every helper:
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"}
if account_id not in user.get("allowed_accounts", []):
return {"error": "forbidden"}
return {"account_id": account_id, "owner": user["user_id"]}
Microsoft Agent Framework's AG-UI host expects authentication on a request header rather than the runtime properties channel. Pass the token via <CopilotKit headers={...}>:
import { CopilotKit } from "@copilotkit/react-core/v2";
<CopilotKit
runtimeUrl="/api/copilotkit"
headers={{
Authorization: `Bearer ${userToken}`,
}}
>
<YourApp />
</CopilotKit>
Validation lives at the host process level: ASP.NET Core's JwtBearer middleware on the .NET host, FastAPI middleware on the Python host. Either way, the AG-UI endpoint refuses to dispatch the agent until the token is verified — so by the time your tools run, the user identity is already trustworthy.
<Tabs groupId="language_microsoft-agent-framework_agent" items={['.NET', 'Python']} persist> <Tab value=".NET">
using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;
using Microsoft.AspNetCore.Authentication.JwtBearer;
using OpenAI;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddAuthentication(JwtBearerDefaults.AuthenticationScheme)
.AddJwtBearer(options =>
{
options.Authority = builder.Configuration["JwtAuthority"];
options.Audience = builder.Configuration["JwtAudience"];
options.TokenValidationParameters = new Microsoft.IdentityModel.Tokens.TokenValidationParameters
{
ValidateIssuer = true,
ValidateAudience = true,
ValidateLifetime = true,
ValidateIssuerSigningKey = true,
};
});
builder.Services.AddAuthorization();
var app = builder.Build();
app.UseAuthentication();
app.UseAuthorization();
string githubToken = builder.Configuration["GitHubToken"]!;
var openAI = new OpenAIClient(
new System.ClientModel.ApiKeyCredential(githubToken),
new OpenAIClientOptions { Endpoint = new Uri("https://models.inference.ai.azure.com") }
);
var agent = openAI.GetChatClient("gpt-5.4-mini")
.CreateAIAgent(name: "AGUIAssistant", instructions: "You are a helpful assistant.");
app.MapAGUI("/", agent).RequireAuthorization();
await app.RunAsync();
Settings live in appsettings.json:
{
"JwtAuthority": "https://login.microsoftonline.com/{your-tenant-id}/v2.0",
"JwtAudience": "api://{your-client-id}",
"GitHubToken": "your-github-token-here"
}
from fastapi import FastAPI, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint
from agent import create_agent
import os
app = FastAPI(title="CopilotKit + Microsoft Agent Framework")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
REQUIRED_BEARER_TOKEN = os.getenv("AUTH_BEARER_TOKEN")
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
if REQUIRED_BEARER_TOKEN and request.url.path == "/":
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")
token = auth_header.split(" ", 1)[1].strip()
if token != REQUIRED_BEARER_TOKEN:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
return await call_next(request)
chat_client = build_chat_client() # Azure OpenAI or OpenAI
my_agent = create_agent(chat_client)
add_agent_framework_fastapi_endpoint(app=app, agent=my_agent, path="/")
Settings live in agent/.env:
AUTH_BEARER_TOKEN=super-secret-demo-token
The most common reason to wire auth is so individual tools can decline to run. Read the resolved user inside the tool's handler and bail if the role doesn't match:
def delete_record(record_id: str, *, user: User):
if "admin" not in user.permissions:
raise PermissionError("admin role required")
# do the delete
This composes with Human in the loop: gate on auth first, surface a confirmation card next, execute last.
Verifying who the caller is doesn't yet stop them reaching someone else's conversation. How much of that you have to build depends on which runtime you're running.
| Runtime | Who scopes threads to a user |
|---|---|
CopilotRuntime with intelligence | Mostly the runtime, via identifyUser — with three routes you still have to guard. |
| Anything else — SSE runtime, custom store, local in-memory runner | You do. See Scope threads yourself. |
The Intelligence runtime requires an identifyUser callback — construction throws without one (or without at least one Channel). It runs on the server, once per request, and the id it returns is the scope the runtime hands to the platform:
const runtime = new CopilotRuntime({
agents: { default: agent },
intelligence,
identifyUser: async (request) => {
const session = await verifyAppSession(request); // Your server-side auth.
if (!session?.user) throw new Error("Unauthorized"); // Backstop; see below.
return { id: session.user.id, name: session.user.name };
},
});
Where a route does resolve the caller, what matters next is whether that id is actually carried to the platform as a scope:
| Route | Scoped to the resolved user? |
|---|---|
agent/run, agent/connect | Yes |
threads/list | Yes — and filtered by agentId, so useThreads returns the caller's threads rather than the project's |
threads/messages | Yes |
threads/update (rename via PATCH, delete via DELETE), threads/archive | Yes |
| Thread subscription token | Yes |
threads/events, threads/state | No — resolves the caller, then ignores it |
agent/stop | No — never resolves a caller at all |
agent/stop never resolves a caller. It goes straight to runner.stop({ threadId }),
which aborts whichever run the runtime is tracking under that thread id. Any caller who
learns an active threadId can kill that run mid-flight.
Guard all three yourself. The onBeforeHandler pattern below applies on the Intelligence
path too, narrowed to these routes:
onBeforeHandler: async ({ request, route }) => {
// Switch rather than an array `includes`, so `route` narrows and
// `route.threadId` type-checks — all three variants carry one.
switch (route.method) {
case "threads/events":
case "threads/state":
case "agent/stop":
break;
default:
return;
}
// None of these is scoped platform-side; check your own ownership record.
const user = await verifyRequest(request);
if (!(await userOwnsThread(user.id, route.threadId))) {
throw new Response("Not found", { status: 404 });
}
},
Without an ownership record of your own, reject these routes outright rather than leaving them open. </Callout>
For everything in the Yes rows there is no ownership table to build. What you own is identifyUser itself:
Request; whatever it returns is trusted from there on. Reading a user id straight out of a header or request body hands every caller the ability to name themselves.401 from onRequest. Beyond the status codes being wrong — an identifyUser that throws surfaces as a 500, a malformed id as a 400 — routes like agent/stop never call it, so a check placed only here isn't reached on every request. Authenticate in onRequest, which runs on every route and can throw a Response directly, and keep the throw inside identifyUser as a backstop.See Scope Rich Threads to the signed-in user for the full runtime contract.
Without the Intelligence Platform there is no server-side binding between a threadId and a user. A threadId is just an opaque id travelling in a request: if user A learns user B's, every thread route accepts it. The rest of this section is the pattern for a custom store, a plain SSE runtime, or the local in-memory runner — and it's also what you narrow to threads/events, threads/state, and agent/stop if you are on the platform.
Whatever stores your threads, keep a record of who each one belongs to. The minimum is a table your runtime can query:
create table thread_owners (
thread_id text primary key,
user_id text not null
);
create index on thread_owners (user_id);
Write a row when a conversation is first created — see minting a thread with your own API for where that hooks into the chat lifecycle.
onBeforeHandleronRequest runs before routing, so it can't see which thread is being addressed. onBeforeHandler runs after, and receives a route that names the operation and — for thread-scoped routes — the threadId:
const handler = createCopilotRuntimeHandler({
runtime,
basePath: "/api/copilotkit",
hooks: {
onRequest: async ({ request }) => {
// Authenticate first: reject anonymous callers outright.
const user = await verifyRequest(request);
if (!user) throw new Response("Unauthorized", { status: 401 });
},
onBeforeHandler: async ({ request, route }) => {
const user = await verifyRequest(request);
// Routes that name a thread directly.
if ("threadId" in route) {
if (!(await userOwnsThread(user.id, route.threadId))) {
throw new Response("Forbidden", { status: 403 });
}
return;
}
// agent/run and agent/connect carry the thread in the body instead.
if (route.method === "agent/run" || route.method === "agent/connect") {
// Clone: the handler still needs to read the original body.
const { threadId } = await request.clone().json();
if (threadId && !(await userOwnsThread(user.id, threadId))) {
throw new Response("Forbidden", { status: 403 });
}
}
},
},
});
The routes that carry a threadId on route are agent/stop, threads/update, threads/archive, threads/messages, threads/events, and threads/state.
threads/list has no threadId to check, so onBeforeHandler has nothing to authorize against. Off the platform the route returns whatever the configured store holds — the local in-memory runner filters by agentId only, and a custom store returns exactly what you wrote. Build the list from your own ownership table instead, and drive the chat with the selected id.
Filter server-side. Hiding rows in the UI leaves the underlying route open.
<Callout type="info"> With no platform thread store, the ownership table above *is* your thread list. See [Self-managed thread persistence](/threads-self-managed). </Callout>onRequest. It is the only hook that runs on every route. identifyUser names a caller; it does not gate one, and some routes never invoke it.identifyUser, plus your own guard on threads/events, threads/state, and agent/stop; off it, an ownership check on every thread route rather than only at login. See Thread authorization.anonymous) instead.