Skip to main content

AAuth Quickstart — LangChain / LangGraph

Wire the AAuth protocol into a LangGraph ReAct agent so every tool call carries cryptographic proof-of-identity. The agent's identity is an Ed25519 key pair registered with LumoAuth; the agent gets a short-lived access token via the AAuth token endpoint; every outgoing tool call is signed with RFC 9421 HTTP Message Signatures (per-request, replay-proof proof of possession) and sent with an Authorization: Bearer <token> header.

Prerequisites
  • LumoAuth organization with AAuth enabled
  • Python 3.9+

Complete Steps 1–3 first (generate a key pair, register the agent and the resource), then verify with the Python SDK quickstart before wiring LangGraph.

Install​

pip install lumoauth[aauth] langchain-openai langgraph

Example​

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from lumoauth.aauth import AAuthClient

# 1. Initialize the AAuth client (load existing key from file in production)
private_pem, jwks = AAuthClient.generate_keypair()
aauth = AAuthClient(
    agent_identifier="https://my-agent.example.com",
    private_key_pem=private_pem,
    org_id="acme-corp",
)

# 2. Obtain a resource token (in production, fetched at task start)
resource_token = "..."  # from your resource server

# 3. Request authorization upfront
tokens = aauth.request_authorization(resource_token=resource_token, scope="read write")
if tokens.get("authorization_required"):
    raise RuntimeError("User consent required - implement OAuth redirect flow first")

ACCESS_TOKEN = tokens["access_token"]

# 4. Define tools that use the AAuth-issued access token
@tool
def fetch_protected_data(endpoint: str) -> str:
    """Fetch data from the protected API using an AAuth-signed request."""
    resp = aauth.signed_request(
        "GET",
        f"https://api.example.com{endpoint}",
        auth_token=ACCESS_TOKEN,
    )
    if resp.status_code == 200:
        return resp.text
    return f"Error: HTTP {resp.status_code}"

@tool
def submit_report(payload: str) -> str:
    """Submit a report to the protected API."""
    resp = aauth.signed_request(
        "POST",
        "https://api.example.com/v1/reports",
        auth_token=ACCESS_TOKEN,
        json={"data": payload},
    )
    if resp.status_code in (200, 201):
        return "Report submitted successfully."
    return f"Error: HTTP {resp.status_code}"

# 5. Create and run the LangGraph ReAct agent
tools = [fetch_protected_data, submit_report]
llm = ChatOpenAI(model="gpt-4", temperature=0)
agent_executor = create_react_agent(llm, tools)

def run(query: str):
    events = agent_executor.stream(
        {"messages": [("user", query)]},
        stream_mode="values",
    )
    for event in events:
        event["messages"][-1].pretty_print()

if __name__ == "__main__":
    run("Fetch /v1/reports/q4-2024 and then submit a summary as a new report.")

How it works​

ComponentRole
AAuthClientManages key material, signs requests, and handles token exchange
aauth.signed_request(...)Wraps requests and adds an RFC 9421 Agent-Auth signature header
ACCESS_TOKENShort-lived token issued by LumoAuth after agent + resource auth
LangGraph ReAct loopLets the LLM decide which tools to call; AAuth handles auth transparently

Next steps​