langchain-runcycles 0.1.1

LangChain agent middleware for Cycles — pre-tool-call authorization, fan-out caps, and per-tenant budget enforcement for Python agents using create_agent.

LangChain Runcycles — AI agent middleware for budget and action authority

LangChain agent middleware for AI agent governance — enforce cost limits, tool permissions, and multi-tenant policies in create_agent workflows before LLM calls or tool actions execute. Works with LangGraph, LangSmith, OpenAI, Anthropic, MCP servers, and any LangChain 1.x agent runtime — built on the new AgentMiddleware API (wrap_tool_call, before_model, wrap_model_call).

AgentMiddleware subclasses for the Cycles Protocol: gate every tool call with wrap_tool_call, cap model fan-out with before_model + jump_to: "end", reserve and commit budget per call — with sync and async support, typed configuration, and optional remote policy decisions. Install via pip install langchain-runcycles.

What's in the box

• CyclesToolGate — runs before every tool call. Authorizes via client.decide() and/or reserves budget via client.create_reservation(). Returns a ToolMessage on denial so the model can recover gracefully.
• CyclesFanOutGate — runs before every model turn. Halts the agent (with jump_to: "end") when a turn cap is hit or when an external policy says to stop. Useful for runaway-loop protection and per-tenant burst caps.

Both work with sync or async LangChain agents and the sync (CyclesClient) or async (AsyncCyclesClient) Cycles client.

Installation

pip install langchain-runcycles

Requires Python 3.10+ and langchain >= 1.0.

Quick Start

from langchain.agents import create_agent
from langchain_core.tools import tool
from langchain_runcycles import CyclesToolGate
from runcycles import Action, CyclesClient, CyclesConfig, Subject

@tool
def send_email(to: str, body: str) -> str:
    """Send an email."""
    return f"Sent to {to}"

client = CyclesClient(CyclesConfig(base_url="http://localhost:7878", api_key="..."))
gate = CyclesToolGate(
    client,
    subject=Subject(tenant="acme", agent="researcher"),
    action={"send_email": Action(kind="tool.call", name="send_email")},
    mode="decide",
)

agent = create_agent(model="claude-sonnet-4-6", tools=[send_email], middleware=[gate])
agent.invoke({"messages": [{"role": "user", "content": "Email alice."}]})

If client.decide() denies the call, send_email is never invoked — the model receives a ToolMessage with the denial reason and can choose another path.

Middleware

CyclesToolGate

Gates each tool call. Three modes:

Mode	What it does
"decide"	Calls client.decide(). Denies the tool call on a non-allow decision. No reservation.
"reserve"	Creates a reservation, runs the tool, commits on success / releases on exception.
"decide+reserve"	Authorizes via decide(), then reserves+commits. Most strict.

gate = CyclesToolGate(
    client,
    subject=Subject(tenant="acme", agent="researcher"),
    action={
        "search": Action(kind="tool.call", name="search"),
        "send_email": Action(kind="tool.call", name="send_email"),
    },
    mode="decide+reserve",
)

CyclesFanOutGate

Halts the agent when a turn cap or external policy says stop. Optional client argument enables remote policy checks on each turn:

from langchain_runcycles import CyclesFanOutGate

fanout = CyclesFanOutGate(
    max_turns=20,
    client=client,                       # optional — for remote policy
    subject=Subject(tenant="acme"),
    action=Action(kind="model.turn", name="research"),
)

Pair with CyclesToolGate and HumanInTheLoopMiddleware for production-grade agent governance.

Configuration

Subject

Either a static Subject or a callable resolving from request/state:

from runcycles import Subject

# Static
subject = Subject(tenant="acme", agent="bot")

# Per-call extractor (CyclesToolGate: (request, state); CyclesFanOutGate: (state, state))
def per_tenant(request, state):
    return Subject(tenant=state["config"]["tenant"], agent="bot")

Action

Static, mapping (per-tool name), or callable:

from runcycles import Action

# Static
action = Action(kind="tool.call", name="any")

# Per-tool mapping
action = {
    "send_email": Action(kind="tool.call", name="send_email"),
    "search": Action(kind="tool.call", name="search"),
}

# Callable
def derive(request):
    return Action(kind="tool.call", name=request.tool_call["name"])

Denial messages

denial_message accepts a format string (placeholders: {reason}, {tool}, {decision}) or a callable receiving the CyclesResponse:

gate = CyclesToolGate(
    client,
    subject=...,
    action=...,
    denial_message="Cycles denied {tool}: {reason}",
)

Error handling

• Denied tool calls return a ToolMessage with the denial content; the underlying handler is never invoked. The agent's model sees the denial as if a tool returned an error and can recover.
• Reservation failures in "reserve" mode are returned as ToolMessage (handler not invoked).
• Tool exceptions in "reserve" mode trigger an automatic release_reservation, then the exception propagates.
• Async/sync mismatch raises TypeError — pair CyclesClient with .invoke() and AsyncCyclesClient with .ainvoke().

Async support

Async middleware variants run automatically when the LangChain agent is invoked with .ainvoke(). Pass an AsyncCyclesClient:

from runcycles import AsyncCyclesClient

async_client = AsyncCyclesClient(CyclesConfig(...))
gate = CyclesToolGate(async_client, subject=..., action=..., mode="decide")

agent = create_agent(model="...", tools=[...], middleware=[gate])
await agent.ainvoke({"messages": [...]})

Examples

• examples/tenant_budget_agent.py — single-tenant budget gate with risky-tool denial recovery.
• examples/multi_agent_fanout.py — multi-agent / HITL flow with CyclesToolGate + CyclesFanOutGate + HumanInTheLoopMiddleware.

Known limitations (v0.1)

• Reserve mode commits at the configured estimate, not actual usage. mode="reserve" and mode="decide+reserve" reserve the estimate, run the tool, then commit the same amount on success. Per-tool actual-cost instrumentation (analogous to runcycles.stream_reservation's cost_fn) is on the roadmap. Until then, set estimate to the worst-case spend per call you're willing to debit, or use mode="decide" if you only want policy gating without budget movement.
• No streaming-LLM cost integration yet. wrap_model_call is not implemented in v0.1; for token-level streaming budget tracking, use runcycles.stream_reservation directly inside an LLM-spend handler.
• Per-call subject only via the extractor form. Static Subject plays one tenant per middleware instance. For per-tenant/per-agent routing in a multi-tenant deployment, supply a SubjectExtractor callable.
• Synthetic tool_call_id when missing. If a ToolCallRequest arrives without an id, the middleware fabricates missing-<hex> for the ToolMessage and logs a warning. Correct LangChain runtimes always supply id; this is a defensive fallback.

Development

pip install -e ".[dev]"
pytest                          # all tests
pytest --cov=langchain_runcycles  # with coverage (gate: ≥95%)
ruff check . && ruff format
mypy langchain_runcycles

Documentation

• LangChain integration page: https://docs.langchain.com/oss/python/integrations/middleware/runcycles (pending PR review)
• Cycles protocol & SDK: https://runcycles.io
• Architecture: see AUDIT.md

Requirements

• Python 3.10+
• runcycles >= 0.4.1
• langchain >= 1.0, < 2.0
• langchain-core >= 0.3

License

Apache-2.0. See LICENSE.