antraft 2.4.2

The governed agent framework — build any agent, with security, scale, and speed by default

Antraft

The agent framework that lets you build any agent imaginable — with security, scale, and speed engineered in from the first line.

Antraft is a framework for building AI agents (like LangChain / LangGraph) with one difference those frameworks structurally cannot match: every action an agent takes is policy-enforced, audited, and governable by default. Governance isn't a plugin you bolt on — it's the substrate every agent runs on.

Build agents from LLMs and tools, compose them into multi-agent graphs, and plug in any MCP server, REST/OpenAPI API, or Anthropic Skill as a governed tool — in Python or JavaScript/TypeScript.

Antraft is also the reference implementation of the MI9 Agent Intelligence Protocol.

---

Why Antraft

	LangChain / LangGraph	Antraft
Build LLM agents	✅	✅
Multi-agent graphs	✅	✅
Parallel tool execution	✅	✅ (each call still governed)
Streaming	✅	✅
Structured / typed output	✅	✅ (pydantic or JSON Schema)
Retrieval / RAG	✅	✅ (as a governed tool)
Providers	many	all of them — Anthropic, OpenAI, Gemini, Bedrock, Azure, Vertex, Cohere, Mistral, Groq, Together, DeepSeek, xAI, Ollama/vLLM… via native adapters + LiteLLM
MCP servers	partial	✅ stdio and remote HTTP/SSE
MCP / API / Skills as tools	partial	✅ unified adapter layer
Policy-enforced by default	❌	✅ every tool call
Audit + replay + PII redaction	❌	✅ built in
Human-in-the-loop pause/approve	❌	✅ built in
Tamper-evident audit (hash-chained)	❌	✅ cryptographically verifiable
RBAC + multi-tenancy	❌	✅ tools scoped by role/tenant
Tool-arg schema validation	❌	✅ rejected at the gateway
Content guardrails (injection/jailbreak/PII/secrets)	❌	✅ inspects what flows through
Observability dashboard (governance-first)	LangSmith (separate)	✅ built in, live
Eval & regression harness	partial	✅ assertions + policy coverage
Long-term memory	✅	✅
Budget governance (token + $ as policy)	❌	✅ deny/pause on spend
Resilient tools (async, retries, timeouts)	partial	✅ built in
Durable checkpoint / resume	LangGraph only	✅ built in
Cross-language governance (Py + JS)	❌	✅ one engine, both languages

Architecture

   Your agent (Python)         Your agent (JS/TS)
          │                          │
   ┌──────▼──────────────────────────▼──────┐
   │  GUARD CONTRACT  (frozen JSON: evaluate)│  ← single source of truth
   ├──────────────────────────────────────────┤
   │  PolicyEngine · Audit · DriftDetector ·  │
   │  SessionRecorder · Swarm control         │
   └──────────────────────────────────────────┘
          ▲                          ▲
   in-process (Python)        HTTP sidecar (any language)

Every tool call — whether from a Python ReActAgent or a TypeScript agent — is evaluated by the same policy engine through one frozen JSON contract. In-process and remote enforcement can never drift.

---

Build a governed agent

import asyncio
from antraft import Antraft, ToolRegistry, tool_from_function
from antraft.models import AnthropicModel          # or OpenAIModel

def search_web(query: str) -> str:
    "Search the web."
    return f"results for {query}"

tools = ToolRegistry([tool_from_function(search_web)])

# Import tools from anywhere — all governed the same way:
#   from antraft import tools_from_mcp_stdio, tools_from_openapi, tools_from_skills_dir
#   tools.extend(tools_from_mcp_stdio("npx", ["-y", "@modelcontextprotocol/server-github"]))

runtime = (
    Antraft.agent(
        model=AnthropicModel(model="claude-opus-4-8"),
        tools=tools,
        task="Research Antraft and summarize it.",
    )
    .allow(["search_web"])
    .record_session("session.jsonl", redact_pii=True)
    .build()
)

asyncio.run(runtime.run())

Any model, any company — one line

from antraft import create_model

m = create_model("anthropic:claude-opus-4-8")          # native adapter
m = create_model("openai:gpt-4o")
m = create_model("gemini:gemini-2.0-flash")
m = create_model("groq:llama-3.1-70b", via="litellm")  # 100+ models via LiteLLM
m = create_model("bedrock/anthropic.claude-3-5-sonnet-20240620-v1:0")  # AWS
m = create_model("azure:my-deployment", via="litellm")                 # Azure
m = create_model("ollama", model="llama3", base_url="http://localhost:11434/v1")  # local

Native adapters for Anthropic / OpenAI / Gemini; everything else (Bedrock, Azure, Vertex, Cohere, Mistral, Groq, Together, DeepSeek, xAI, Replicate, Ollama…) is reachable through the bundled LiteLLM universal adapter — and every one of them is governed identically.

Any tool, anywhere

from antraft import (
    tools_from_mcp_stdio, tools_from_mcp_http,   # MCP: local & remote
    tools_from_openapi, rest_tool,                # REST / OpenAPI
    tools_from_skills_dir,                        # Anthropic Skills
    retriever_tool,                               # RAG
)

tools.extend(tools_from_mcp_stdio("npx", ["-y", "@modelcontextprotocol/server-github"]))
tools.extend(tools_from_mcp_http("https://my-mcp.example.com/mcp"))   # remote MCP

Compose agents into a graph

from antraft.graph import Graph, START, END

g = Graph()
g.add_node("research", run_research_agent)   # each node can run a governed agent
g.add_node("write", run_writer_agent)
g.add_edge(START, "research")
g.add_edge("research", "write")
g.add_conditional_edges("write", lambda s: END if s["ok"] else "research")
result = g.run({"topic": "antraft"})

Govern a JavaScript/TypeScript agent

Run the governance sidecar (Python), then guard JS tool calls with the same engine:

import { GuardClient } from "@antraft/guard";

const guard = new GuardClient({ baseUrl: "http://localhost:8000", policy: { allow: ["search"] } });
await guard.guarded({ name: "search", params: { q: "antraft" } }, () => doSearch("antraft"));

See sdk-ts/ for the TypeScript client.

---

Production-grade by default

runtime = (
    Antraft.agent(model, tools, task="...")
    .allow(["search_web", "send_email"])
    .budget(max_tokens=200_000, max_cost_usd=5.0)   # enforce SPEND as policy
    .resilient(retries=2, timeout=30)                # async tools, retries, timeouts
    .record_session("run.jsonl", redact_pii=True)    # audit + replay + PII scrub
    .checkpoint("run.ckpt")                           # crash-safe, resumable
    .build()
)

# Resume a crashed run with its full state AND remaining budget intact:
runtime = Antraft.resume(
    Antraft.agent(model, tools, task="..."), "run.ckpt"
)

• Budget governance — when an agent crosses its token or dollar limit, Antraft denies (hard) or pauses for approval (soft). Spend is policy, not an afterthought.
• Resilient execution — tools may be sync or async; calls are retried with backoff and bounded by timeouts at the single chokepoint where actions run.
• Durability — every executed action is checkpointed atomically; Antraft.resume continues exactly where a crashed process stopped.

Trust: guardrails, memory, evals, observability

from antraft import Antraft, SemanticMemory

runtime = (Antraft.agent(model, tools, task="...",
                         guardrails=True,                    # injection/jailbreak/PII/secrets
                         memory=SemanticMemory(path="mem.jsonl"))  # learns across runs
    .observe()        # live dashboard at /observability
    .build())

• Guardrails inspect content, not just actions: prompt-injection in tool results is withheld, jailbreak/secret content is blocked, PII is redacted.
• Memory recalls relevant facts from past runs and stores new outcomes.
• Observability streams every allow/deny/pause, spend, and guardrail block to a live dashboard (uvicorn antraft.api.server:app → /observability).

Test agents before shipping:

from antraft.eval import run_eval, EvalCase, assert_tool_called, assert_final_contains, assert_no_denials

report = run_eval(
    EvalCase("math", "what is 17+25", [
        assert_tool_called("add"), assert_final_contains("42"), assert_no_denials(),
    ]),
    build=lambda task: Antraft.agent(model, tools, task=task).allow(["add"]),
)
print(report.summary())   # pass/fail + policy coverage (allow/deny/pause counts)

Proof, not just claims

Reproducible benchmarks (no API keys needed):

python benchmarks/run.py

Representative numbers (your hardware will vary):

Metric	Result
Governance overhead	~16 µs/eval (~60k evals/sec)
Schema validation	+~28 µs/call
Parallel tool speedup	~2.4× (4 tools)
Guardrail scan	~50 µs/scan

Guardrail effectiveness is measured and enforced in CI by an adversarial red-team suite (tests/test_guardrail_redteam.py): injection recall ≥ 80%, jailbreak ≥ 85%, secret leakage 0 misses, benign false-blocks 0%.

Govern an existing agent (no rewrite)

Already have an agent? Wrap it — Antraft stays framework-agnostic (LangChain, AutoGen, CrewAI, custom).

---

Key Capabilities

1. Robust Governance

Secure your agents with deterministic rules.

• Policy Enforcement: Approve, deny, or pause every action before execution.
• Stateful Control: Enforce prerequisites (e.g., "Must authenticate before reading data").
• Rich Logic Rules: define complex constraints like amount > 1000 and not user_verified.
• Path Confinement: Restrict file access to specific directories.

2. Deep Observability

Complete visibility into agent behavior.

• Session Recording: Capture full execution traces (Inputs, Thoughts, Actions, Outputs) for replay debugging.
• PII Redaction: Automatically scrub sensitive data (Emails, SSNs) from logs.
• Cognitive Telemetry: Log the agent's reasoning process, not just its actions.
• Drift Detection: Identify anomalous behavior patterns (loops, spikes) in real-time.

4. Distributed Control

Manage fleets of agents at scale.

• Swarm Protocol: Control thousands of agents across different servers from a central API.
• Human-in-the-Loop: Pause high-risk actions and wait for human approval via API.
• Dynamic Policies: Update agent permissions on-the-fly without restarting.

5. Universal Compatibility (MCP)

Antraft runs as a Model Context Protocol (MCP) server, making it instantly compatible with:

• Claude Desktop
• Cursor / Windsurf IDEs
• Any MCP-compliant client

---

Installation

pip install antraft

---

Quick Start ("Fluent" SDK)

The Antraft SDK provides a fluent interface to secure ANY python agent in seconds.

import asyncio
from antraft import Antraft

# 1. Define your Agent & Tools
agent = MyAgent()
tools = {
    "search": google_search_tool,
    "shell": run_shell_tool
}

async def main():
    # 2. Wrap & Secure
    await Antraft.guard(agent, tools) \
        .allow(["search"]) \
        .deny(["shell"]) \
        .record_session("logs/session_01.jsonl", redact_pii=True) \
        .run()

if __name__ == "__main__":
    asyncio.run(main())

---

Advanced Usage

Defining Complex Policies (YAML)

For enterprise use cases, load policies from file.

# policy.yaml
allow:
  - read_file
rules:
  - trigger: "action:transfer_money"
    checks:
      - "amount > 10000"
    enforce: "pause"
    message: "High value transfer requires approval."

await Antraft.guard(agent, tools) \
    .load_policy("policy.yaml") \
    .run()

Swarm Mode (Remote Control)

Connect an agent to a central Control Plane.

1. Start the Server:

python -m antraft.cli.main serve

2. connect Agents:

await Antraft.guard(agent, tools) \
    .connect_swarm("http://localhost:8000") \
    .run()

---

Documentation

• User Guide: Comprehensive "How-To".
• Architecture: System design and components.
• Threat Model: Security analysis.
• Syntax Guide: Configuration reference.

Example Project

• python examples/refund_agent/app.py: realistic support-refund workflow with HITL approval and session recording.

Development

python -m pytest -q

Contributing

We welcome contributions! Please see CONTRIBUTING.md for details.

License

MIT License