authgraph

The DAG framework where every agent action needs a permit before it runs.

pip install authgraph

from authgraph import AGEFRuntime

authgraph is a Python DAG execution framework where every node transition asks a TrigGuard gateway "is this allowed?" and refuses to proceed without a PERMIT receipt. No permit → no execution. Every action leaves a signed receipt.

Powered by TrigGuard. Requires a TrigGuard gateway (hosted free tier: 5,000 decisions/month, no credit card).

Previously known internally as AGEF (Authority-Governed Execution Framework).

---

What authgraph is

authgraph is a DAG-based execution orchestration framework for AI agents. It answers a question that Temporal, Airflow, LangGraph, and CrewAI don't:

Before this agent executes this task — is it actually authorized to?

authgraph wraps every claim, every execution, and every completion in a TrigGuard authorization call. The result is a verifiable audit trail of receipts that proves every action was permitted before it happened.

---

Relationship to TrigGuard

┌─────────────────────────────────────────────────┐
│                    authgraph                        │
│  DAG engine  ·  Coordinator  ·  Verification    │
│                                                 │
│           AuthorityGate (gate.py)               │
│    ↓ authorize(surface, actor, context) ↓       │
└──────────────────┬──────────────────────────────┘
                   │  public SDK only
                   ▼
┌─────────────────────────────────────────────────┐
│                  TrigGuard                      │
│  Authorization  ·  Receipts  ·  Policy          │
└─────────────────────────────────────────────────┘

authgraph is a consumer of TrigGuard. It calls TrigGuard's public authorize() SDK method and stores the returned receipt IDs as references. authgraph never:

• Imports TrigGuard internals
• Mints or signs receipts
• Bypasses the authority boundary
• Modifies TrigGuard policy

The AuthorityGate class (gate.py) is the only file in authgraph that touches TrigGuard. Every other module is framework-neutral.

---

Architecture

AGEFRuntime
├── ExecutionGraphEngine    DAG validation, Kahn's topological sort, state machine
├── AuthorityGate           Single TrigGuard integration point
├── Coordinator             Claim leases, TTL, duplicate suppression
├── VerificationEngine      Evidence checks, hallucination detection
└── OutcomeTracker          Drift scoring, policy flywheel data

Node lifecycle

PENDING → READY → CLAIMED → EXECUTING → VERIFYING → COMPLETE
                                                   ↘ FAILED → READY (retry)

Every PENDING → CLAIMED and VERIFYING → COMPLETE transition requires a TrigGuard PERMIT.

TrigGuard surfaces

Surface	When
planning.task.create	Before READY → CLAIMED
artifact.publish	Before VERIFYING → COMPLETE
execution.external_action	Before releasing downstream external-effect nodes

---

Quick start

import os
from authgraph import AGEFRuntime, EvidenceArtifact, VerificationTier
import hashlib

# Set your TrigGuard API key
os.environ["TRIGGUARD_API_KEY"] = "your-key"

runtime = AGEFRuntime()

graph = runtime.create_graph(title="Deploy workflow", description="Plan → Build → Release")

plan    = runtime.add_node(graph, title="Plan",    description="Define scope")
build   = runtime.add_node(graph, title="Build",   description="Implement",
                            verification_tier=VerificationTier.HIGH)
release = runtime.add_node(graph, title="Release", description="Publish",
                            authority_surface="artifact.publish",
                            verification_tier=VerificationTier.CRITICAL)

runtime.add_dependency(graph, from_node=plan, to_node=build, transfer_outputs=["spec"])
runtime.add_dependency(graph, from_node=build, to_node=release, transfer_outputs=["artifact"])
runtime.start(graph)

# Claim and execute each node
for node in [plan, build, release]:
    claim = runtime.attempt_claim(graph, node, agent_id="my-agent")
    runtime.execute_node(graph, node, claim_token=claim["claim_token"])

    content = f"work done on {node.title}"
    h = hashlib.sha256(content.encode()).hexdigest()
    runtime.submit_completion(
        graph, node,
        claim_token=claim["claim_token"],
        outputs={"result": f"{node.title} complete"},
        evidence=[EvidenceArtifact(
            artifact_id=f"ev-{h[:8]}", artifact_type="TEXT",
            content_hash=h, content_ref=content, produced_by="my-agent",
        )],
    )

print(graph.status)  # COMPLETE

---

CrewAI example

from authgraph import AGEFRuntime, VerificationTier

runtime = AGEFRuntime()
graph = runtime.create_graph(title="Research crew", description="Authority-governed research")

research = runtime.add_node(graph, title="Research", description="Market analysis")
report   = runtime.add_node(graph, title="Report",   description="Publish findings",
                              authority_surface="artifact.publish")

runtime.add_dependency(graph, from_node=research, to_node=report)
runtime.start(graph)

# Wrap each CrewAI task execution with attempt_claim / execute_node / submit_completion
# TrigGuard receipt is stored on node.complete_authority_receipt_id

See examples/crewai_example.py for the full pattern.

---

LangGraph example

bridge = AuthgraphLangGraphBridge()
bridge.register_node("planner", "Generate plan")
bridge.register_node("executor", "Execute plan", depends_on=["planner"])
bridge.start()

@bridge.guarded_node("planner")
def planner_node(state):
    return {**state, "plan": "do the thing"}

Each @bridge.guarded_node decorator wraps the LangGraph node function in an authgraph authority gate. The TrigGuard receipt is stored in the LangGraph state under agef_receipts.

See examples/langgraph_example.py for the full pattern.

---

Quanta example

bridge = AuthgraphQuantaBridge()
graph = bridge.mission_to_graph(quanta_mission)

result = bridge.run_task(graph, "Engineering Build", eng_agent,
                          outputs={"artifact": "v2.tar.gz"},
                          work_summary="Build complete")

Quanta creates missions; authgraph executes them with authority gates. Quanta is a consumer of authgraph — any other orchestrator can replace it.

See examples/quanta_example.py for the full pattern.

---

Authorization example

# gate.py handles all authorization — nothing else in authgraph touches TrigGuard
from authgraph import AuthorityGate

gate = AuthorityGate()  # reads TRIGGUARD_API_KEY from environment

decision = gate.authorize_claim(
    node_id="node-abc",
    graph_id="graph-xyz",
    agent_id="my-agent",
    surface="planning.task.create",
    context={"title": "Research task"},
)

if decision.permitted:
    print(f"PERMIT — receipt: {decision.receipt_id}")
else:
    print(f"DENIED — {decision.reason}")  # fail-closed

If TrigGuard is unreachable, permitted is False and the node stays READY. Nothing proceeds without a receipt.

---

Outcome tracking example

from authgraph import OutcomeTracker, OutcomeClassification

tracker = OutcomeTracker()

tracker.record(
    node,
    execution_summary="Agent completed research on schedule",
    classification=OutcomeClassification.CORRECT,
    execution_drift_score=0.05,
    outcome_drift_score=0.10,
)

print(tracker.summary())
# {"total": 1, "by_classification": {"CORRECT": 1}, "avg_execution_drift": 0.05, "success_rate": 1.0}

# Export for policy model training
records = tracker.training_records()

---

Environment variables

Variable	Default	Description
TRIGGUARD_API_KEY	—	TrigGuard API key (required for live authorization)
TRIGGUARD_GATEWAY_URL	https://api.trigguardai.com	TrigGuard gateway URL

---

Install

pip install authgraph

from authgraph import AGEFRuntime, ExecutionGraph, ExecutionNode, NodeStatus

For development:

git clone https://github.com/TrigGuard-AI/agef
cd agef
pip install -e ".[dev]"
pytest tests/

---

Authority boundary guarantee

# Zero TrigGuard internal imports in any authgraph module except gate.py
grep -r "from trigguard\." authgraph/ | grep -v gate.py  # → 0 matches

# Zero receipt minting
grep -r "mint_receipt\|receipt_chain" authgraph/          # → 0 matches

# authorize() called only in gate.py
grep -rn "\.authorize(" authgraph/ | grep -v gate.py      # → 0 matches

---

License

MIT — see LICENSE