Metadata-Version: 2.4
Name: authsec-langchain-sdk
Version: 0.1.2
Summary: AuthSec identity, delegation, and CIBA approval for LangChain agents
Author-email: AuthSec <support@authsec.ai>
License: Apache-2.0
Project-URL: Homepage, https://authsec.ai
Project-URL: Documentation, https://docs.authsec.ai
Project-URL: Source, https://github.com/authsec-ai/authsec-langchain
Keywords: langchain,authsec,auth,oauth,spiffe,delegation,ciba,agents
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2.5
Requires-Dist: langchain-core>=0.2
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Requires-Dist: respx>=0.21; extra == "dev"
Requires-Dist: ruff>=0.5; extra == "dev"
Requires-Dist: mypy>=1.10; extra == "dev"
Provides-Extra: examples
Requires-Dist: langchain>=0.2; extra == "examples"
Requires-Dist: langchain-openai>=0.1; extra == "examples"

# 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.

[![PyPI](https://img.shields.io/pypi/v/authsec-langchain-sdk)](https://pypi.org/project/authsec-langchain-sdk/)
[![Python](https://img.shields.io/pypi/pyversions/authsec-langchain-sdk)](https://pypi.org/project/authsec-langchain-sdk/)

---

## Why this exists

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

```python
# 😱 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

```bash
pip install authsec-langchain-sdk
```

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

```python
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)

```python
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:

```python
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:

```python
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:

```python
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`)

```python
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="admin@example.com",
            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

```python
@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

```python
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

```python
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":       "delegator@example.com",
    "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`:

```python
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):

```python
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`):

```python
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`](examples/smoke_local.py) | Pulls the delegation token, checks cache, force-refreshes. No agent, no LLM. Good for confirming setup. |
| [`examples/real_integration.py`](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`](examples/aws_s3_agent.py) | Full LangChain agent + OpenAI + real AWS S3 via SPIRE-exchanged STS. |

Run any of them:

```bash
$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

```bash
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
