Add JWT authentication

Set "auth": "jwt" in 10xgraph.json, put JWT_SECRET_KEY and JWT_ALGORITHM in your environment, and send Authorization: Bearer <token> on every request. The token must carry exp and user_id claims. The user_id becomes the thread owner and reaches your tools through the run config.

How does a request carry the token?

The server reads a standard bearer credential from the Authorization header. Browser and server clients send the same header.

To pin tokens to your service, set JWT_ISSUER and JWT_AUDIENCE. When either is set, tokens must carry a matching claim, so a token minted for another service that shares the key is rejected.

How does the user reach my tools?

The server copies the verified identity into the run config as config["user"] (the full decoded claims) and config["user_id"]. A tool receives that config by declaring a parameter named config. Framework-injected parameters are hidden from the tool schema the model sees.

def list_my_orders(config: dict) -> list[dict]:  
    """List the caller's recent orders."""
    user_id = config["user_id"]  
    return db.orders_for(user_id)

Never take the user id from a tool argument. The model chooses tool arguments, and a prompt-injected model could ask for another user’s data.

How do I make users see only their own threads?

Set "authorization": "ownership". A thread is read, continued, stopped or fixed only by the user who created it. Another user’s request on that thread is denied for every action, including invoke and stream. Listing threads is scoped to the caller.

If you leave authorization unset, the default depends on MODE: ownership when MODE=production, and allow-all otherwise. Setting it explicitly keeps development and production consistent.

Ownership is looked up through the checkpointer you pass to compile(). Use a durable one such as PgCheckpointer so owners survive restarts. Set redis in 10xgraph.json (or REDIS_URL) to share the ownership cache across workers.

For roles and scopes, authorization also accepts an object with "backend": "rbac", a roles map and optional default_scopes. See the configuration reference.

How do I write custom auth instead?

Subclass BaseAuth and return a dict containing at least user_id. Raise HTTPException(status_code=401) for a bad credential.

from agentflow_cli import BaseAuth
from fastapi import HTTPException

class AgentAuth(BaseAuth):
    def authenticate(self, request, response, credential):
        if credential is None:
            raise HTTPException(status_code=401, detail="Missing credentials")
        claims = verify_with_my_idp(credential.credentials)  # raises on a bad token
        return {"user_id": claims["sub"], "roles": claims.get("roles", [])}

Point the config at it with the object form:

{
  "auth": { "method": "custom", "path": "auth.agent_auth:AgentAuth" }
}

Every key you return is merged into config["user"]. Derive user_id only from the verified credential, never from a client-set header.