LangGraph

LangGraph (and LangChain more broadly) gets a governed ChatOpenAI pointed at your Agent Fabric LLM proxy. LangGraph is the deep adapter: every proxy header and the SDK’s shared async transport reach the native client, and the adapter runs the full conformance suite in CI.

What you get

• A native langchain_openai.ChatOpenAI — nothing LangGraph-specific wraps it.
• Per-run correlation IDs that reach every graph node.
• Typed gateway refusals (PIIDetected, TokenBudgetExceeded, …) inside nodes.
• A conformance suite you can run against your own graph.

Install

pip install "donkey-kit[langgraph]"

Quickstart

from donkey_kit.integrations.langgraph import chat_model
 
llm = chat_model("gpt-4o")

import OpenAI from "openai";
 
const client = new OpenAI({
  baseURL: process.env.DONKEY_LLM_PROXY_URL,   // no /v1
  apiKey: "unused",                              // required slot; proxy uses the headers below
  defaultHeaders: {
    client_id: process.env.DONKEY_LLM_PROXY_CLIENT_ID!,
    client_secret: process.env.DONKEY_LLM_PROXY_CLIENT_SECRET!,
  },
});
 
const reply = await client.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Say hi in three words." }],
});
console.log(reply.choices[0].message.content);

Three ways to construct

1. Off a shared Donkey instance (reuses one HTTP client and lifecycle across every adapter you use in a run):

from donkey_kit import Donkey
 
async with Donkey.from_env() as donkey:
    llm = donkey.langgraph.chat_model("gpt-4o")

The adapter is also callable: donkey.langgraph("gpt-4o") is the same as donkey.langgraph.chat_model("gpt-4o").

2. Module-level factory (shortest — uses a cached, env-configured default Donkey):

from donkey_kit.integrations.langgraph import chat_model
 
llm = chat_model("gpt-4o")

3. Governed kwargs, native constructor (you call ChatOpenAI yourself):

from donkey_kit import Donkey
from langchain_openai import ChatOpenAI
 
async with Donkey.from_env() as donkey:
    llm = ChatOpenAI(model="gpt-4o", **donkey.langgraph.connection_kwargs())

Manual equivalent

The factories make this native constructor call for you:

from langchain_openai import ChatOpenAI
 
llm = ChatOpenAI(
    model="gpt-4o",
    base_url=...,             # from DONKEY_LLM_PROXY_URL, no /v1 suffix
    api_key=...,
    default_headers=...,      # client_id / client_secret header pair, not bearer
    http_async_client=...,    # the SDK's shared httpx async client
    max_retries=0,            # the SDK retries in its own transport layer
    use_responses_api=True,   # the proxy's data plane is /responses
)

connection_kwargs() returns exactly these keys, so you can drop the factory and construct ChatOpenAI by hand at any time. Pass use_responses_api=False to chat_model(...) only if your deployment exposes chat-completions instead of the Responses API (/responses).

Graph-level features

Correlation IDs reach every node

Bind a run ID once with donkey.run(id=…) and every node sees it via current_correlation_id(), with nothing threaded through graph state. LangGraph runs nodes on asyncio tasks that copy the current context, so the ID propagates on its own:

from donkey_kit.core.telemetry import current_correlation_id
 
async def prepare(state):
    logger.info("handling", extra={"correlation_id": current_correlation_id()})
    return {}
 
async with donkey.run(id=ticket.id):
    await graph.ainvoke({"messages": [("user", ticket.text)]})

Typed refusals inside a node

On a proxy refusal, LangChain raises its own wrapped OpenAIPermissionDeniedError, not the SDK’s typed exception. Wrap the model call in typed_refusals() and a gateway refusal comes back through the error taxonomy instead:

from donkey_kit.integrations.langgraph import typed_refusals
 
async def call_model(state):
    with donkey.langgraph.typed_refusals():   # or: with typed_refusals():
        reply = await model.ainvoke(state["messages"])
    return {"messages": [reply]}

A PII block now propagates out of graph.ainvoke(...) as PIIDetected, a budget block as TokenBudgetExceeded, and so on — each carrying the correlation and call IDs the client sent. Transport-level errors with no HTTP response (connection failures, timeouts) pass through unchanged.

interrupt() composes with typed refusals

A human-in-the-loop interrupt() and a typed refusal don’t interfere: the graph pauses cleanly at the interrupt, and on resume a refusal in a downstream model node still surfaces as its typed exception.

Run the conformance suite against your own graph

The suite that tests this adapter is also a pytest plugin you can point at your own agent:

pytest --donkey-conformance --agent=my_app:build

build returns an object with an awaitable run(text). The suite checks that it doesn’t retry a budget refusal, surfaces PIIDetected typed, carries the correlation ID into its logs, and tolerates a response with no budget headers. The examples/langgraph factory has exactly this shape. See Testing.

Notes

• base_url, api_key, default_headers, and a custom http_async_client are all forwarded, so proxy auth headers and the SDK’s transport (retries, correlation IDs) reach every request.
• max_retries=0 is intentional: retries live in the SDK’s transport layer, so the SDK and the OpenAI client don’t both retry.
• The proxy is OpenAI-compatible but not the full OpenAI API: the base URL has no /v1 prefix, there is no /models endpoint, and auth is a client_id/client_secret header pair rather than a bearer token.

See the error taxonomy for how proxy rejections surface as typed exceptions.