(|=|) Fences

The open-source SDK for Fences, on-call for AI agents. It records what your agent decides and why, enforces limits the agent can't raise, and lets you stop a run from the Fences dashboard. Works offline too: local limits, no account.

import agentfences
from agentfences import governed, checkpoint, log_decision

agentfences.init(local_only=True)

@governed(budget_usd=0.50, max_iterations=20, max_tokens=50_000)
async def run_agent(query: str, messages: list):
    while True:
        log_decision(reasoning="calling LLM", action="llm_call")
        response = call_llm(messages)

        result = await checkpoint(response)  # Fences reads the usage and prices the call

        if result.breached:
            # Agent handles the limit gracefully — no exception, no crash
            messages.append({"role": "system", "content": result.system_prompt})
            return call_llm(messages)  # LLM summarises what it found

        if response.is_done():
            return response

Python and TypeScript

This README covers the Python SDK (pip install agentfences). There's also a TypeScript SDK for Node 18+ (npm install agentfences, see sdk-js). Both talk to Fences the same way (PROTOCOL.md).

What each part guarantees, and what it can't, is in ENFORCEMENT.md.

Install

pip install agentfences

How it works

checkpoint() returns a CheckpointResult after every call — not an exception. The agent reads it and decides what to do: stop, summarise, ask for more budget. No try/catch in your orchestration layer.

result = await checkpoint(response)

result.ok           # True if within all limits
result.breached     # True if a limit was crossed
result.breach_type  # "budget_exceeded" | "iteration_limit" | "time_limit" | "token_limit"
result.message      # "I've reached my budget limit... I'll summarise what I found."
result.system_prompt  # ready to inject into LLM conversation context
result.spent_usd    # the run's total spend so far

Cost tracking

Pass the model's response to checkpoint() and Fences works out what the call cost. It reads the model name and token usage, including cache reads and writes and reasoning tokens, and prices them from a built-in table of current list prices:

response = client.chat.completions.create(model="gpt-4o", messages=messages)
result = await checkpoint(response)
print(result.spent_usd)  # the run's total so far

• Supported responses: OpenAI (Chat Completions and Responses), Anthropic, Gemini and LangChain messages, or any object or dict with the same usage fields. If a gateway such as OpenRouter reports the billed cost, that is used instead.
• Built-in prices: about 290 models from OpenAI, Anthropic, Google, Mistral, DeepSeek, xAI and Groq, including long-context pricing. They come from LiteLLM's price list and are refreshed each release.
• Other models: a model missing from the table still has its tokens counted, and Fences logs a warning once, because budget_usd can't stop a model it can't price. Add prices in USD per 1M tokens:

  agentfences.init(local_only=True, prices={"my-finetune": {"input": 2.0, "output": 6.0}})
  

• Extra costs: add a paid tool call or anything else to the step with checkpoint(response, cost_delta_usd=0.01). Without a response, cost_delta_usd is the step's whole cost.
• Streaming: pass the final message, or the last chunk once it carries usage (OpenAI: stream_options={"include_usage": True}).

Quickstart — no backend needed

import asyncio
import agentfences
from agentfences import governed, checkpoint, log_decision

agentfences.init(local_only=True)  # no account, no API key

@governed(budget_usd=0.10, max_iterations=5, max_tokens=2000)
async def my_agent(query: str):
    for step in range(100):
        log_decision(reasoning=f"step {step}: searching", action="search")

        result = await checkpoint(cost_delta_usd=0.02, tokens_used=200)
        if result.breached:
            return result.message  # or inject result.system_prompt into your LLM

    return "done"

print(asyncio.run(my_agent("hello")))
# I've reached my budget limit (spent $0.1200 of $0.1000). I'll summarize what I found so far.

Limits

A limit trips once it is exceeded, not when it is reached: a $0.10 budget allows exactly $0.10 of spend.

Every limit is optional. On Fences Cloud, a limit left out of @governed(...) comes from the agent's page in the dashboard, so @governed() on its own works. Limits set nowhere default to 100 steps, 5 minutes and no token limit, and no budget: the run still starts, but its spend isn't capped, and Fences logs a warning once per agent (and flags the agent in the dashboard).

On Fences Cloud a run can also stop with stopped_by_user (someone pressed Stop in the dashboard), paused (paused and not resumed in time), key_daily_budget or key_monthly_budget (the agent's daily or monthly cap across all its runs). With init(fail_closed=True), an unreachable backend stops it with fences_unreachable. Each comes with its own message and system_prompt.

Sync agents and streaming

@governed works on sync functions, async functions and async generators. Sync agents use checkpoint_sync(), which takes the same arguments and returns the same CheckpointResult:

from agentfences import governed, checkpoint_sync

@governed(budget_usd=0.50)
def my_agent(query: str):
    while True:
        response = call_llm(query)
        result = checkpoint_sync(response)
        if result.breached:
            return result.message

A governed async generator (a streaming agent) stays governed until the stream is exhausted. Concurrent agents in one event loop and nested @governed calls each get their own run.

Frameworks: LangChain, LangGraph and the OpenAI Agents SDK

Agents built on a framework need no decorators. Each model call is prechecked against the budget before it's sent (see "Budgets are prechecked" below; set max_tokens on the model for small budgets), then recorded with its tokens and cost and checkpointed, so limits apply, and each tool call is recorded. Limits work as in @governed and are all optional.

Callbacks and hooks can't hand a result back to the agent, so when a limit is crossed these integrations raise FencesStop. It carries the same breach_type, message and system_prompt as a breached checkpoint().

LangChain and LangGraph (needs langchain-core):

from agentfences import FencesStop
from agentfences.langchain import FencesCallbackHandler

try:
    graph.invoke(inputs, config={"callbacks": [FencesCallbackHandler(budget_usd=2)]})
except FencesStop as stop:
    answer = stop.message

The outermost run becomes a Fences run named after it (or agent_name=...), and joins the run if it's already inside @governed. Sync and async graphs both work. LangChain logs its own "Error in FencesCallbackHandler.on_llm_end callback" line when the handler stops a graph; that's expected.

OpenAI Agents SDK (needs openai-agents, Python 3.10+):

from agentfences import FencesStop, openai_agents

try:
    result = await openai_agents.run(agent, "Research cats", budget_usd=2)   # Runner.run, governed
except FencesStop as stop:
    answer = stop.message

Already inside @governed? Use Runner.run(agent, input, hooks=openai_agents.FencesRunHooks()).

With a framework integration, don't also pass instrument=True or decorate tools with @agentfences.tool: each call would be recorded twice.

Live control

In cloud mode the SDK keeps a heartbeat with Fences, so whoever is on call can act on a run while it's going:

• Stop: the run's next checkpoint() returns breach_type="stopped_by_user" with its message, so the agent wraps up gracefully. A stop sticks, whatever later responses say. If an async run doesn't reach a checkpoint within stop_grace_s (default 30 seconds) of the stop, because it's stuck in a tool say, it's cancelled and the governed call raises FencesStop. init(stop_grace_s=None) turns that off. A synchronous run can't be interrupted, so it stops at its next checkpoint or guarded tool.
• Pause and resume: a paused run waits inside its next checkpoint() until it's resumed, then carries on. If no one resumes it within pause_timeout_s (default one hour), the checkpoint returns breach_type="paused".
• Change limits: new limits apply to the run immediately.
• Quarantine: calling a governed function for a quarantined agent raises AgentQuarantined before any of its code runs. Its runs already going hold at their next checkpoint until the quarantine is released, and their guarded tools are blocked.

An agent can also ask a person before doing something risky:

approval = await agentfences.request_approval("Refund $240 to order 1182?", amount_usd=240, timeout_s=600)
if not approval.granted:
    return f"I didn't issue the refund: {approval.note}"

The answer has granted, by, note and amount_usd; an approved amount raises the run's budget by that much. request_approval_sync() does the same for synchronous agents. It never hangs and never raises: in local mode, outside a run, or when the server can't take requests it returns denied right away, and with no answer it returns denied after timeout_s.

Containment hooks run in your agent's process when a stop or quarantine arrives, to contain what Fences can't reach, such as a credential the agent holds:

def revoke(event):  # {"type": "quarantine", "run_id", "agent_name", "id", "by", "reason"}
    vault.revoke(f"agent/{event['agent_name']}")

agentfences.init(api_key="fc_...", on_quarantine=revoke, on_stop=lambda event: ...)

Each runs in a background thread: on_stop once per run, on_quarantine once per quarantine however many runs it reaches. Fences records whether the hook worked, raised (and why) or ran past 30 seconds, on the incident's timeline. A process that exits right after a stop waits for its hooks to finish first, up to 30 seconds; in serverless handlers, flush() does the same.

Acknowledgements: the SDK confirms every dashboard command it receives, so the dashboard can tell "the agent has it" from "the agent hasn't acknowledged". A process that's dead, cut off or stuck in synchronous code gets flagged within about 30 seconds.

The heartbeat reports live runs every 15 seconds (init(heartbeat_s=...)), and every 2 seconds while a run is paused or waiting for approval. A Fences server without these features gets one warning: stops still apply at the next checkpoint, and approval requests are denied.

Guarded tools

A checkpoint runs after a step, so on its own it notices a limit once it's crossed. For actions with consequences (refunds, payments, writes to production), guard the tool, and it asks Fences before it runs:

@agentfences.tool(cost_usd=lambda order_id, amount: amount, approve=True)
def issue_refund(order_id: str, amount: float): ...

@agentfences.tool(guard=True, timeout_s=30)
async def deploy(service: str): ...

• The tool runs only if its run isn't stopped, paused, over a limit or quarantined. Otherwise it raises ToolBlocked (with reason and breach_type) and its code never runs. The model can't talk its way past this: the check is in the tool itself.
• cost_usd (a number, or a function of the tool's arguments) is reserved before the call and charged after, so five parallel $40 refunds against a $100 budget can't all go through: two do. A call that raises is released uncharged. One that times out is charged, because it may have happened.
• approve=True asks a person on the dashboard before each call (see request_approval above). The dashboard can also require approval for a tool, and the SDK then asks for it.
• Guards fail closed: if Fences can't be reached, or doesn't support guarded tools yet, the tool is blocked. In local mode the SDK applies the same rules itself, and approve=True is always denied, as there's no one to ask.

timeout_s works on any tool, guarded or not: past it, the call raises TimeoutError. An async tool is cancelled; a sync one can't be, so it finishes in a background thread while the agent moves on.

Evidence: tool calls and model calls

On-call can only explain what was recorded. Fences records two kinds of call on the run automatically, alongside your log_decision() entries.

Tool calls. Decorate your agent's tools:

@agentfences.tool
def web_search(query: str, limit: int = 5) -> list: ...

@agentfences.tool(name="fetch")
async def fetch_page(url: str) -> str: ...

Each call records the tool's name, a short summary of its arguments (query='cats', limit=5), whether it succeeded (and the error if not) and how long it took. Exceptions still reach your code, and outside a governed run the tool is just called.

Model calls. Pass instrument=True to record every OpenAI and Anthropic call made inside a governed run, and precheck it against the budget, with no other code changes:

agentfences.init(api_key="fc_...", endpoint="https://...", instrument=True)

Each call records the model, tokens (including cache reads and writes), cost, latency, and the error and status code if it failed (a 429 rate limit, say). It doesn't count as a checkpoint, so keep calling checkpoint(response) where limits should be checked. Streamed calls are recorded without token counts.

Budgets are prechecked. In a run with a budget, each instrumented call, and each call made through a framework integration, is checked before it's sent. If its worst-case cost could take the run past the budget, it raises FencesStop (budget_exceeded) and never reaches the provider:

try:
    resp = client.chat.completions.create(model="gpt-4o", messages=msgs, max_tokens=1000)
except agentfences.FencesStop as stop:
    return stop.message  # "I can't make another model call without risking going over my budget ..."

The worst case is the request's input, estimated high, plus its output cap at the output rate. The cap is max_tokens if you set it, otherwise the model's whole output limit (16,384 tokens for gpt-4o), so set max_tokens on calls in runs with small budgets. Each call's worst case is held while it's in flight, so parallel calls can't overspend together. Calls you make without instrumentation are only checked afterwards, by checkpoint(). See ENFORCEMENT.md.

Both kinds of call are recorded as spans named by the OpenTelemetry GenAI conventions (chat gpt-4o, execute_tool web_search, with gen_ai.* attributes plus Fences' fences.cost_usd). They're kept on the run (get_active_run().events, newest 1,000) and, in cloud mode, sent to Fences in the background in batches. They pass through the redact hook first as tool_call (name, args, error) and llm_call (model, error) events. A Fences server that doesn't accept them yet gets a single warning and nothing else changes.

Context, errors and redaction

Tag runs so on-call can see which deploy and which customer a run belongs to:

agentfences.init(api_key="fc_...", endpoint="https://...", environment="prod", release="v1.4.2")

with agentfences.context(user_id="u_123", session_id="s_9", trace_id=span_id):
    await support_agent(question)   # every run started in here carries these values

Blocks nest (inner values win), and get_active_run().context shows a run's values from inside the agent.

When a governed run raises, Fences records the exception type, message and the last 15 stack frames with the run, and its error reads ValueError: page 7 returned 403 rather than just the message. The exception still reaches your code as usual.

Everything that leaves your process (a run's context, decisions, and a failed run's error) passes through an optional redact hook first. Return the event, changed or not, or None to drop it:

def redact(event):
    if event["type"] == "decision":
        event["reasoning"] = EMAIL.sub("[email]", event["reasoning"])
    return event

agentfences.init(api_key="fc_...", endpoint="https://...", redact=redact)

Events have a type of run_start (context), decision (reasoning, action), guard (tool, arguments of a guarded call: dropping it still lets the check happen, without arguments), span (a tool or model call: change its attributes, e.g. gen_ai.tool.call.arguments or fences.error.message) or run_end (status, error, exception). If the hook raises, that event is dropped rather than sent unredacted, and a warning is logged once. Runs still start and end, so limits keep working.

Modes

Local (free) — governance runs entirely in-process. No account, no backend, no API key.

agentfences.init(local_only=True)

Cloud (coming soon) — persistent audit trails, live dashboard, server-authoritative enforcement. Same code, one line changes.

agentfences.init(api_key="fc_...", endpoint="https://your-fences-instance.com")

In cloud mode the backend is authoritative: it sees spend from every process sharing a run, so one process can't overspend because it didn't see another's spend. On Fences Cloud, a limit raised from the dashboard lets the run's next checkpoint() pass, so a fenced agent can resume. Backend calls run off the event loop and log_decision() never blocks; decisions are sent in the background and flushed at exit (call agentfences.flush() yourself before a serverless handler returns).

If the backend can't be reached, limits are enforced locally and a warning is logged once per run. To treat an unreachable backend as a breach instead:

agentfences.init(api_key="fc_...", endpoint="https://...", fail_closed=True)

Project layout

fences/
├── docs/       landing page
├── sdk/        agentfences Python package
└── sdk-js/     agentfences TypeScript package (npm)
    └── agentfences/

License

MIT