authsec-langchain-sdk 0.1.2

AuthSec identity, delegation, and CIBA approval for LangChain agents

authsec-langchain-sdk

AuthSec identity, delegation, and human-in-the-loop approval for LangChain agents.

Give your LangChain agents scoped, short-lived, audited identities instead of static cloud keys. Three lines of integration code; the rest is policy you define in AuthSec.

---

Why this exists

The common pattern for LangChain agents that touch real cloud APIs:

# 😱 don't do this in production
os.environ["AWS_ACCESS_KEY_ID"] = "AKIA..."
os.environ["AWS_SECRET_ACCESS_KEY"] = "..."
agent.invoke("delete all my buckets")

Every problem with that approach:

Problem	What AuthSec does instead
Long-lived static keys in env vars	Short-lived JWT, auto-refreshed
No human-in-the-loop authority	JWT carries the delegating user's user_id and email
Agent has whatever the IAM user has	Policy intersects user perms ∩ allowed perms
No expiry	TTL capped at the policy level (e.g. 1h max)
Can't tell who did what	Every issuance + use is auditable by client_id, user_id, spiffe_id

The SDK is the agent-side glue. AuthSec is the policy engine.

---

Install

pip install authsec-langchain-sdk

The PyPI name uses dashes; the Python import name uses underscores:

from authsec_langchain import AuthsecClient, AuthsecConfig, AuthsecCallbackHandler

---

Prerequisites — one-time setup in AuthSec

Before any code runs, an AuthSec admin needs to do this for each agent (via the AuthSec UI or curl):

1. Register the AI agent as a client (client_type=ai_agent, agent_type=langchain) → gets a client_id UUID
2. Provision a SPIFFE identity for the agent → writes spiffe_id to the client record
3. Create a delegation policy for (role_name, agent_type=langchain, allowed_permissions, max_ttl_seconds)
4. Delegate a token to the agent → writes a row to delegation_tokens with status=active

After step 4, the agent can pull its token via this SDK indefinitely (until the row is revoked or expires).

The agent's developer only needs two things from this setup: the base_url of AuthSec and the agent's client_id. No JWTs, secrets, or roles handed to the agent.

---

Quick start (4 lines)

from authsec_langchain import AuthsecClient, AuthsecConfig

client = AuthsecClient(AuthsecConfig(
    base_url="https://auth.example.com",
    client_id="a594430b-2bd4-4792-9666-63162ee858c5",
))

# Returns the current delegation JWT for this agent.
# Cached in-memory; auto-refreshes near expiry.
token = client.get_delegation_token()

That's the core. Everything else in this guide builds on this one call.

---

Integration patterns

Pattern A — Use the token directly in a LangChain @tool

The agent's tools call AuthSec for identity, then forward the JWT to whatever downstream service:

import httpx
from langchain_core.tools import tool
from authsec_langchain import AuthsecClient, AuthsecConfig

authsec = AuthsecClient(AuthsecConfig(
    base_url="https://auth.example.com",
    client_id="a594430b-2bd4-4792-9666-63162ee858c5",
))

@tool
def query_billing_api(customer_id: str) -> dict:
    """Look up a customer's billing details."""
    jwt = authsec.get_delegation_token()
    resp = httpx.get(
        f"https://billing.internal/customers/{customer_id}",
        headers={"Authorization": f"Bearer {jwt}"},
    )
    resp.raise_for_status()
    return resp.json()

The downstream service verifies the JWT (signed by AuthSec, public keys at /.well-known/jwks.json) and reads the permissions claim to decide what's allowed.

Pattern B — Drop the callback into an AgentExecutor

The AuthsecCallbackHandler makes the current JWT available on the run's metadata so any tool can read it without needing the client directly:

from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_openai import ChatOpenAI
from authsec_langchain import AuthsecClient, AuthsecConfig, AuthsecCallbackHandler

authsec = AuthsecClient(AuthsecConfig(
    base_url="https://auth.example.com",
    client_id="a594430b-2bd4-4792-9666-63162ee858c5",
))

agent = create_tool_calling_agent(ChatOpenAI(model="gpt-4o-mini"), tools=[...], prompt=...)

executor = AgentExecutor(
    agent=agent,
    tools=[...],
    callbacks=[AuthsecCallbackHandler(authsec)],
)

executor.invoke({"input": "What's customer ACME's plan?"})

The callback refreshes the token before every tool invocation. If the token is expired, get_delegation_token() round-trips to AuthSec; otherwise it's served from cache.

Pattern C — Trade the JWT for AWS / Azure / GCP credentials

When the agent needs to call cloud APIs directly with their native SDKs (boto3, azure-sdk, google-cloud-*), exchange the delegation JWT for cloud-native credentials:

import boto3

# Returns AWS STS credentials issued under your AuthSec OIDC trust.
creds = authsec.exchange_cloud_credentials(
    "aws",
    audience="sts.amazonaws.com",
    role_arn="arn:aws:iam::123456789012:role/agent-s3-reader",
)

session = boto3.Session(
    aws_access_key_id=creds["access_key_id"],
    aws_secret_access_key=creds["secret_access_key"],
    aws_session_token=creds["session_token"],
    region_name="us-east-1",
)

session.client("s3").list_buckets()

Identical pattern for Azure ("azure") and GCP ("gcp"). The cloud STS service handles federation; the agent never sees long-lived cloud keys.

Pattern D — Pause for human approval (CIBA)

High-risk actions can require human approval. The agent's tool calls request_approval(), which:

1. POSTs to AuthSec's CIBA initiate endpoint
2. The user gets a push notification on their device
3. The SDK polls until the user taps approve/deny
4. Returns an approval JWT (or raises CIBADeniedError / CIBATimeoutError)

from authsec_langchain import CIBADeniedError, CIBATimeoutError

@tool
def delete_customer_data(customer_id: str) -> str:
    """Permanently delete a customer's data. REQUIRES HUMAN APPROVAL."""
    try:
        approval_jwt = authsec.request_approval(
            login_hint="[e-mail]",
            binding_message=f"Allow agent to delete customer {customer_id}?",
            requested_expiry=300,  # 5 minutes
        )
    except CIBADeniedError:
        return "Action denied by user."
    except CIBATimeoutError:
        return "Approval timed out; not proceeding."

    # Proceed with deletion using the approval_jwt as auth
    ...

---

Configuration

@dataclass(frozen=True)
class AuthsecConfig:
    base_url: str               # required — AuthSec server root
    client_id: str              # required — your agent's UUID, registered in AuthSec
    api_token: Optional[str]    # optional — only for admin endpoints
    tenant_id: Optional[str]    # optional — uses tenant CIBA endpoints when set
    timeout: float = 10.0
    ciba_poll_interval: float = 2.0
    ciba_poll_timeout: float = 120.0

Recommended: pull config from env

import os

authsec = AuthsecClient(AuthsecConfig(
    base_url=os.environ["AUTHSEC_BASE_URL"],
    client_id=os.environ["AUTHSEC_AGENT_CLIENT_ID"],
    tenant_id=os.environ.get("AUTHSEC_TENANT_ID"),
))

Setting these in env vars (or your secrets manager) means the agent code is identical across dev / staging / prod — only config changes.

---

What's in the JWT

token = client.get_delegation_token()
# Decoded (base64 of the middle segment):
{
    "sub":         "spiffe://example.org/tenants/<tenant>/agents/langchain/<client_id>",
    "iss":         "spiffe://<tenant-uuid>",
    "aud":         ["authsec-api"],
    "agent_type":  "langchain",
    "client_id":   "a594430b-...",
    "user_id":     "<the human who delegated>",
    "email":       "[e-mail]",
    "permissions": ["s3:read", "billing:read"],
    "exp":         1779261936,
    "iat":         1779258336,
    "jti":         "<unique token id>"
}

Downstream services verify the signature against AuthSec's public JWKS, then make authorization decisions based on the permissions claim and any other policy logic they have.

---

Error handling

All SDK errors inherit from AuthsecError:

from authsec_langchain import (
    AuthsecError,        # base class
    DelegationError,     # token fetch / cloud exchange failed
    CIBARequiredError,   # a downstream action needs approval
    CIBADeniedError,     # user denied an approval request
    CIBATimeoutError,    # approval polling timed out
)

try:
    jwt = client.get_delegation_token()
except DelegationError as e:
    # 400 / 404 / 410 / network errors — log + fail the tool call
    log.error("AuthSec delegation failed: %s", e)
    raise

Common cases:

Error	Cause	Fix
DelegationError: ... 400	Missing or malformed client_id	Check config
DelegationError: ... 404 No active delegation token	Admin hasn't called /delegate-token for this agent yet	Run the prereq setup
DelegationError: ... 410	Token row exists but is expired	Admin needs to re-delegate
CIBADeniedError	User denied the approval prompt	Tool should fail gracefully
CIBATimeoutError	No response within ciba_poll_timeout (default 120s)	Increase timeout or fall back to deny

---

Caching behavior

get_delegation_token() caches the JWT in-memory with a 60-second safety margin before expiry. So if your token expires at 12:00:00, the SDK will fetch a fresh one starting at 11:59:00.

To force a refresh (e.g. you know the policy just changed):

fresh = client.get_delegation_token(force_refresh=True)

For high-throughput agents, the cache means AuthSec gets one request per token lifetime — not one per tool call.

---

Verifying the JWT downstream

The cleanest path for downstream services (Go / Python / Node / etc.):

1. Fetch AuthSec's JWKS once at startup: GET {AUTHSEC_BASE_URL}/authsec/.well-known/jwks.json
2. Cache the key set; refresh on JWT kid miss
3. On each request, verify the JWT signature with the matching key, check exp, check aud, then make authz decisions from permissions

Example (Python with pyjwt):

import jwt as pyjwt
import httpx

JWKS = httpx.get("https://auth.example.com/authsec/.well-known/jwks.json").json()

def verify_agent_jwt(token: str) -> dict:
    headers = pyjwt.get_unverified_header(token)
    key = next(k for k in JWKS["keys"] if k["kid"] == headers["kid"])
    public_key = pyjwt.algorithms.RSAAlgorithm.from_jwk(key)
    return pyjwt.decode(token, public_key, algorithms=["RS256"], audience="authsec-api")

---

Examples in this repo

File	What it does
examples/smoke_local.py	Pulls the delegation token, checks cache, force-refreshes. No agent, no LLM. Good for confirming setup.
examples/real_integration.py	Self-contained mock-API demo — agent fetches token, calls a "downstream service", which checks permissions. No cloud setup needed.
examples/aws_s3_agent.py	Full LangChain agent + OpenAI + real AWS S3 via SPIRE-exchanged STS.

Run any of them:

$env:AUTHSEC_BASE_URL = "https://auth.example.com"
$env:AUTHSEC_AGENT_CLIENT_ID = "<your agent UUID>"
python examples/smoke_local.py

---

Common pitfalls

"Where do I get the client_id from?" The AuthSec admin gets it when registering the agent via POST /authsec/clientms/tenants/:tenantId/clients/create with client_type=ai_agent. The response includes the client_id. Hand that UUID to the developer; no other credentials needed.

"The token expired mid-request — does the SDK auto-retry?" Not yet (v0.1). If you hit a 401 from a downstream service due to expiry, call get_delegation_token(force_refresh=True) and retry. Auto-retry on 401 is on the roadmap.

"Can I use this without LangChain?" Yes. The AuthsecClient class is framework-neutral — only AuthsecCallbackHandler depends on LangChain. Import just the client if you're integrating with a different framework or a plain Python tool.

"Async support?" v0.1 is sync only. AsyncAuthsecClient is coming in v0.2. For now, wrap calls in asyncio.to_thread() if you need them in an async context.

---

Status

Feature	Status
Delegation-token fetch + cache	✅
AWS / Azure / GCP cloud exchange	✅
CIBA initiate + poll (sync)	✅
LangChain callback handler	✅
Async client (AsyncAuthsecClient)	⏳ v0.2
LangGraph node helpers	⏳ v0.2
Auto-retry on downstream 401	⏳ v0.2
Webhook-based CIBA	⏳ v0.3

---

Contributing

git clone https://github.com/authsec-ai/authsec-langchain
cd authsec-langchain
pip install -e ".[dev]"
pytest
ruff check .

Tests are mocked; no live AuthSec needed for pytest. To run integration tests against a real AuthSec, set AUTHSEC_BASE_URL and AUTHSEC_AGENT_CLIENT_ID.

---

License

Apache 2.0