Authentication

By default, the 10xGraph API accepts all requests without authentication (suitable for local development). For production, configure either JWT or a custom auth backend.

No authentication (default)

{
  "agent": "graph.react:app",
  "auth": null
}

All endpoints are publicly accessible. Only use this locally or behind a secure gateway.

JWT authentication

Set auth to "jwt" in 10xgraph.json:

{
  "agent": "graph.react:app",
  "auth": "jwt"
}

JWT support needs an extra:

pip install "10xscale-agentflow-cli[jwt]"

Set the required environment variables:

JWT_SECRET_KEY=your-secret-key
JWT_ALGORITHM=HS256

Both are validated when the config loads: "auth": "jwt" with either missing raises a ValueError and the server does not start.

Requests must include a Bearer token in the Authorization header:

curl -X POST http://127.0.0.1:8000/v1/graph/invoke \
  -H "Authorization: Bearer eyJhbGciOiJIUzI1..." \
  -H "Content-Type: application/json" \
  -d '{"messages": [...], "config": {"thread_id": "t1"}}'

The server decodes and validates the JWT on every request. The decoded payload becomes the user context, and flows into config["user"] for every downstream call.

Required claims

The built-in backend rejects a token that does not carry both of these:

Everything else in the payload is passed through untouched, so roles, scopes, email, and any custom claims reach your tools and authorization backend as-is.

Rejection codes

Failures raise UserAccountError, which the error handler returns as HTTP 403 with the code in the body.

On a WebSocket route these become a clean close with code 1008 instead of an HTTP response.

Generating tokens

10xGraph does not issue tokens. Use your identity provider, or mint one for testing. The payload must carry user_id and exp:

import datetime

import jwt  # PyJWT, installed by the [jwt] extra

token = jwt.encode(
    {
        "user_id": "user-123",
        "exp": datetime.datetime.now(datetime.UTC) + datetime.timedelta(hours=1),
        "roles": ["member"],
    },
    key="your-secret-key",
    algorithm="HS256",
)

A token keyed on sub instead of user_id is rejected with INVALID_TOKEN, and a token with no exp is rejected even if it is otherwise valid.

Sending the token over WebSocket

WebSocket routes use the same bearer credential as HTTP, but browsers cannot set an Authorization header on a WebSocket handshake. The server therefore looks in three places, in this order:

1. Authorization: Bearer <token> — non-browser clients.
2. Sec-WebSocket-Protocol: agentflow-bearer, <token> — the preferred browser mechanism. The token rides in a request header, so it never lands in a URL, an access log, or browser history. The server echoes the agentflow-bearer sentinel back on accept(), which browsers require in order to complete the handshake.
3. ?token=<jwt> — a last-resort fallback, accepted on WebSocket connections only. The token ends up in URLs and access logs.

const ws = new WebSocket(
  "ws://localhost:8000/v1/graph/ws",
  ["agentflow-bearer", token],  // sentinel first, then the raw JWT
);

The offer must be exactly two entries with the sentinel first. Anything else falls through to the query parameter.

Custom authentication

Provide your own auth backend when you need integration with an internal identity system:

{
  "agent": "graph.react:app",
  "auth": {
    "method": "custom",
    "path": "graph.auth:MyAuthBackend"
  }
}

The backend must subclass BaseAuth and implement authenticate. The method is synchronous and receives three arguments: the connection, the response, and the bearer credential (or None).

from typing import Any

from fastapi import Request, Response
from fastapi.security import HTTPAuthorizationCredentials

from agentflow_cli import BaseAuth

class MyAuthBackend(BaseAuth):
    def authenticate(
        self,
        request: Request,          # may be a Request or a WebSocket (HTTPConnection)
        response: Response,
        credential: HTTPAuthorizationCredentials | None,  # bearer token, or None
    ) -> dict[str, Any] | None:
        """Verify the request and return the user context.

        - Return a dict (with at least ``user_id``) to accept. The keys flow into
          ``config["user"]`` for every downstream call.
        - Return ``{}`` / ``None`` for an anonymous request.
        - Raise ``UserAccountError`` (always 403) or ``HTTPException`` (your chosen
          status) to reject. On a WebSocket route either becomes a clean 1008 close.
        """
        if credential is None:
            return None
        claims = my_identity_service.verify(credential.credentials)
        return {
            "user_id": claims["sub"],
            "email": claims.get("email"),
            "roles": claims.get("roles", []),   # consumed by RBAC / custom authorization
            "scopes": claims.get("scopes", []), # consumed by per-endpoint scope checks
        }

Authorization

Authorization decides what an authenticated user may do and whose data they can touch. It is configured separately from auth, via the authorization key.

Built-in backends and mode-based defaults

You do not have to write code to get per-user isolation. Two backends ship built in:

A production build (MODE=production) enforces owner-only access even if you never set authorization. Development stays permissive. The developer’s explicit choice always wins.

{ "agent": "graph.react:app", "auth": "jwt", "authorization": "ownership" }

An unrecognised built-in name raises a ValueError at startup listing the valid ones, so a typo fails loudly rather than silently falling back to something permissive.

The classes behind these names are importable, so you can subclass or compose them:

from agentflow_cli.src.app.core.auth.authorization import (
    AuthorizationBackend,             # the abstract base
    DefaultAuthorizationBackend,      # "allow_all" / "default" / "none"
    OwnershipAuthorizationBackend,    # "ownership"
    RoleBasedAuthorizationBackend,    # the RBAC config block
)

How ownership is resolved

OwnershipAuthorizationBackend only ownership-checks the thread-scoped resources graph and checkpointer. store, files, and config pass through, because they enforce their own per-user scoping.

Its decision table:

ThreadOwnershipResolver

Ownership is immutable: it is set when a thread is created and never changes, only disappearing on delete. That makes it safe to cache aggressively, and ThreadOwnershipResolver does, so an authorization check is not a database round-trip per request.

Negative results are never cached. A None result means the thread does not exist yet, and it could become owned by the very next request. Caching that would let a later caller be treated as the creator of a thread someone else just made.

The L2 tier is wired automatically from redis in 10xgraph.json, falling back to REDIS_URL. When neither is set, or when the redis package is not installed, the resolver runs L1-only and logs a warning at startup.

The eviction contract

Two methods are part of the backend contract, and a custom backend that caches ownership should honour both:

Skipping evict leaves a deleted thread’s owner cached, so the id cannot be reused by a different user until the process restarts.

Custom authorization backend

Subclass AuthorizationBackend. authorize is required; isolation_scope and scopes_for are optional overrides.

from typing import Any

from agentflow_cli.src.app.core.auth.authorization import AuthorizationBackend

class MyAuthorizationBackend(AuthorizationBackend):
    async def authorize(
        self,
        user: dict[str, Any],
        resource: str,          # "graph" | "checkpointer" | "store" | "files" | "config"
        action: str,            # "invoke" | "stream" | "read" | "write" | "delete" | ...
        resource_id: str | None = None,   # thread_id / memory_id when the path carries one
        **context: Any,
    ) -> bool:
        """Return True to allow, False to deny (403)."""
        if not user.get("user_id"):
            return False
        if resource == "store":
            return user.get("role") == "admin"
        return True

    def isolation_scope(self) -> str:
        """How storage partitions rows for this backend:
        "owner" -> scope every checkpointer/store query to the caller; "none" -> no scoping.
        Stamped server-side into config["user"]["authz"]; the client cannot forge it.
        """
        return "owner"

    def scopes_for(self, user: dict[str, Any]) -> list[str] | None:
        """Scopes this identity carries, or None for unrestricted (the permissive default)."""
        return user.get("scopes")

Point 10xgraph.json at it (no method wrapper — just the path):

{ "authorization": "graph.auth:MyAuthorizationBackend" }

Resource and action pairs

The server passes these (resource, action) pairs to authorize. The required scope for an endpoint is "<resource>:<action>".

Role-based access control (no code)

For roles to scopes, use the RBAC config block instead of writing a backend. It loads RoleBasedAuthorizationBackend, which maps roles to scopes on top of the owner-only isolation it inherits from OwnershipAuthorizationBackend:

{
  "authorization": {
    "backend": "rbac",
    "roles": {
      "admin":  ["*"],
      "member": ["graph:invoke", "graph:stream", "graph:read", "checkpointer:read"]
    },
    "default_scopes": ["graph:read"],
    "isolation": "owner"
  }
}

An authorization object whose backend/type is not one of the three RBAC names raises a ValueError at startup.

The same behaviour is available in code when you want to compute the role table at runtime:

from agentflow_cli.src.app.core.auth.authorization import RoleBasedAuthorizationBackend

backend = RoleBasedAuthorizationBackend(
    role_scopes={
        "admin": ["*"],
        "member": ["graph:invoke", "graph:stream", "checkpointer:read"],
    },
    default_scopes=["graph:read"],
    isolation="owner",
)

Point 10xgraph.json at it with "authorization": "module:backend". Subclass it when you need to derive roles from something other than a claim: override scopes_for and everything else, including owner-only thread isolation, still applies.

Scope precedence

A request is allowed only if the identity’s scopes include the endpoint’s "<resource>:<action>". The server resolves that list in a fixed order:

1. authz.scopes_for(user) on the configured authorization backend, when it defines one.
2. user["scopes"] from the authenticated identity, but only when step 1 returned None.

A backend that returns a list wins outright; the identity’s own scopes claim cannot widen it.

The resolved value has three distinct meanings:

The base AuthorizationBackend.scopes_for passes through user["scopes"] when it is a list, tuple, or set, and returns None otherwise, which is why an identity with no scopes claim stays unrestricted.

After a successful check the resolved list is stamped into user["authz"]["scopes"] server-side, so downstream code reads the trusted value rather than anything the client sent.

Data isolation (the config["authz"] contract)

Object-level authorize and scope checks run at the API layer. The data layer (checkpointer, store) enforces isolation separately, driven by a trusted policy: after a successful check the server stamps user["authz"] = {user_id, scope, scopes} where scope comes from the backend’s isolation_scope(). Every service copies that trusted user into config["user"], so the policy reaches the core library and cannot be forged by the client. With scope: "owner" the checkpointer and store scope every row to the caller; with scope: "none" they do not.

The boot-time route guard

Authorization is applied per handler with Depends(RequirePermission(resource, action)). That is precise, but it is also easy to forget on a new endpoint, and a forgotten guard ships an open endpoint. assert_all_routes_protected turns that into a startup failure instead: it runs after the routers are mounted and walks every APIRoute and APIWebSocketRoute, checking the whole dependency subtree for a RequirePermission instance.

If any non-public route lacks one, the server raises and refuses to start:

RuntimeError: Refusing to start: the following routes are not protected by RequirePermission.
Add the dependency, or add the path to the public allowlist if it is intentionally open:
  - POST /v1/my-new-endpoint

The check costs nothing per request. Starlette infrastructure routes (/docs, /redoc, /openapi.json) are not APIRoutes and are skipped automatically.

Public paths

DEFAULT_PUBLIC_PATHS is exactly three entries:

Adding to the allowlist requires editing the frozenset in the source, which is deliberate: it makes opening a route a reviewable change rather than a config edit.

Using auth with the TypeScript client

const client = new AgentFlowClient({
  baseUrl: "http://127.0.0.1:8000",
  headers: {
    Authorization: `Bearer ${token}`,
  },
});

The headers object is merged into every request the client makes.