Authentication

By default, the 10xGraph API accepts all requests without authentication (suitable for local development). For production, configure either JWT or a custom BaseAuth backend to verify identity, and optionally set authorization rules via AuthorizationBackend to control what authenticated users may do.

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"
}

Install the JWT extra:

pip install "10xgraph-api[jwt]"

Set the required environment variables:

JWT_SECRET_KEY=your-secret-key
JWT_ALGORITHM=HS256

Both are validated at startup. Missing either one raises ValueError and the server does not start.

Using JWT

Send 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, available to your graph as config["user"].

Required JWT claims

The built-in backend requires:

If JWT_ISSUER or JWT_AUDIENCE is set, the matching iss or aud claim is also required.

Other claims (roles, scopes, email, custom fields) are passed through to config["user"] unchanged.

Status codes for JWT failures

On WebSocket routes, errors become a clean close with code 1008 instead of an HTTP response.

Generating test tokens

10xGraph does not issue tokens; use your identity provider or mint one locally for testing:

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 using sub instead of user_id is rejected with 401. A token with no exp is also rejected with 401.

Sending auth over WebSocket

WebSocket routes check three places in order for a credential:

1. Authorization: Bearer <token>: for non-browser clients.
2. Sec-WebSocket-Protocol: 10xgraph-bearer, <token>: preferred for browsers (the token stays in headers, never in URLs or logs). The server echoes the sentinel back on accept, which browsers require.
3. ?token=<jwt>: last resort for WebSocket only; the token appears in URLs and access logs.

Example:

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

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

The older sentinel agentflow-bearer is still accepted until 2.0, and the server echoes whichever one you send. The TypeScript client still sends agentflow-bearer.

Custom authentication backend

Provide your own BaseAuth subclass when you need integration with an internal identity system:

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

Subclass BaseAuth and implement authenticate. The method is synchronous and receives the request, response, and credential:

from typing import Any
from fastapi import Request, Response
from fastapi.security import HTTPAuthorizationCredentials
from tenxgraph_api import BaseAuth

class MyAuthBackend(BaseAuth):
    def authenticate(
        self,
        request: Request,  # Request or WebSocket (HTTPConnection)
        response: Response,
        credential: HTTPAuthorizationCredentials | None,  # bearer token or None
    ) -> dict[str, Any] | None:
        """Return user context dict, None for anonymous, or raise to reject.
        
        - Return dict (with at least 'user_id') to authenticate.
        - Return None/{} for anonymous requests.
        - Raise UserAccountError (401) or HTTPException (your status) to reject.
        - On WebSocket: 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", []),
            "scopes": claims.get("scopes", []),
        }

Authorization (access control)

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

Built-in backends

Two backends ship built in; choose one or write your own:

Example:

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

An unrecognized built-in name raises ValueError at startup. The classes are importable for subclassing or composition:

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

Ownership model

OwnershipAuthorizationBackend checks ownership only for thread-scoped resources: graph and checkpointer. The store, files, and config resources enforce their own per-user scoping.

Decision table:

Ownership caching

Ownership is immutable, so it is cached by ThreadOwnershipResolver for scalability:

Negative results are not cached (a missing thread could be created by a later request). The L2 tier is wired from redis in 10xgraph.json, falling back to REDIS_URL. When neither is set or the redis package is not installed, L1 only is used and a warning is logged at startup.

Cache eviction happens on thread delete via CheckpointerService.delete_thread. A custom caching backend should implement evict(thread_id) (returns the resolver’s evict coroutine or None) and aclose() (closes any client on shutdown).

Custom authorization backend

Subclass AuthorizationBackend. The authorize method is required; isolation_scope and scopes_for are optional:

from typing import Any
from tenxgraph_api.src.app.core.auth.authorization import AuthorizationBackend

class MyAuthBackend(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 available
        **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:
        """Storage partitioning: "owner" scopes rows to caller; "none" does not.
        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 (default)."""
        return user.get("scopes")

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

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

Resources and actions

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

Role-based access control (no code)

Use the RBAC config block instead of writing a backend to map roles to scopes. It loads RoleBasedAuthorizationBackend, which inherits owner-only isolation from OwnershipAuthorizationBackend:

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

An unrecognized backend/type value raises ValueError at startup. The same backend is available in code:

from tenxgraph_api.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 compute roles at runtime: override scopes_for, and owner-only isolation still applies.

Scope resolution

A request is allowed only if the identity’s scopes include the endpoint’s "<resource>:<action>". Scopes are resolved in order:

1. authz.scopes_for(user) from your backend (if it defines the method).
2. user["scopes"] from the authenticated identity (only if step 1 returns None).

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

The base AuthorizationBackend.scopes_for passes through user["scopes"] if it is a list/tuple/set, otherwise returns None (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.

Data isolation contract

The API layer enforces object-level authorize and scope checks. The data layer (checkpointer, store) enforces isolation separately via a trusted policy stamped by the server after a successful check: user["authz"] = {user_id, scope, scopes}, where scope comes from isolation_scope(). Every service copies that trusted user into config["user"], so the policy reaches the graph and cannot be forged by the client. With scope: "owner", checkpointer and store queries are scoped to the caller; with scope: "none", they are not.

Boot-time route guard

Authorization is applied per handler with Depends(RequirePermission(resource, action)). To prevent forgotten guards from shipping open endpoints, assert_all_routes_protected runs at startup and walks every APIRoute and APIWebSocketRoute to check that the entire dependency subtree includes a RequirePermission instance.

If any non-public route lacks one, the server 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 is free per request. Starlette infrastructure routes (/docs, /redoc, /openapi.json) are not APIRoutes and are skipped automatically.

Public paths

These three paths are publicly accessible:

Adding paths to the allowlist requires editing the source frozenset, which is deliberate: opening a route becomes a reviewable change.

Using auth with the TypeScript client

import { TenxGraphClient } from "10xgraph-client";

// authToken is sent as a Bearer token on every request
const client = new TenxGraphClient({
  baseUrl: "http://127.0.0.1:8000",
  authToken: token,
});

The token is fixed per client, so create a new client to rotate it. See Create a client for all options.