TealTiger

The ag2.extensions.tealtiger module adds deterministic governance guardrails to AG2 agents. TealTigerMiddleware enforces tool allowlists, detects PII and secrets in tool arguments, tracks per-session cost, provides per-agent kill switches, and produces structured TEEC audit receipts — with no LLM in the governance path.

Installation#

pip install ag2

Import directly:

1
2
3
4
5
6
7

from ag2.extensions.tealtiger import (
    GovernanceDecision,
    GovernanceMode,
    GovernancePolicy,
    TEECReceipt,
    TealTigerMiddleware,
)

Quick Start#

TealTigerMiddleware is a middleware factory — pass the instance straight into an agent's middleware list.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

import asyncio

from ag2 import Agent
from ag2.config import AnthropicConfig
from ag2.extensions.tealtiger import GovernanceMode, GovernancePolicy, TealTigerMiddleware

governance = TealTigerMiddleware(
    mode=GovernanceMode.ENFORCE,
    policies=[
        GovernancePolicy.tool_allowlist(["search", "read_*"]),
        GovernancePolicy.pii_block(["ssn", "credit_card", "email", "phone"]),
        GovernancePolicy.secret_detection(),
        GovernancePolicy.cost_limit(max_per_session=5.0),
    ],
)

async def main() -> None:
    agent = Agent(
        "assistant",
        config=AnthropicConfig(model="claude-sonnet-4-6"),
        middleware=[governance],
    )
    reply = await agent.ask("Summarize the AG2 middleware guide.")
    print(await reply.content())
    print(governance.decisions)
    print(f"Cost: ${governance.total_cost:.4f}")

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

Every tool call now flows through deterministic governance evaluation before execution.

Multi-Stage Defense#

TealTiger governs at two stages of the AG2 agent lifecycle:

flowchart LR
    A[👤 User Message] --> B{🛡️ Stage 1<br/>on_turn}
    B -->|Kill Switch<br/>Check| C[🤖 Agent<br/>Processing]
    C --> D[🔧 Tool Call]
    D --> E{🛡️ Stage 2<br/>on_tool_execution}
    E -->|"✅ ALLOW"| F[⚡ Execute Tool]
    E -->|"❌ DENY"| G[🚫 ToolErrorEvent]
    F --> H[📝 TEEC Receipt]
    G --> H

    style B fill:#f0fdfa,stroke:#14b8a6,stroke-width:2px
    style E fill:#f0fdfa,stroke:#14b8a6,stroke-width:2px
    style H fill:#f0fdfa,stroke:#14b8a6,stroke-width:2px
    style F fill:#f0fdf4,stroke:#16a34a
    style G fill:#fef2f2,stroke:#dc2626

Both stages activate from a single middleware instance. The middleware factory pattern keeps state (decisions, receipts, cost, frozen agents) alive across turns.

Governance Modes#

mode accepts either a GovernanceMode member or its string name, and defaults to GovernanceMode.ENFORCE. Start with MONITOR in staging to see what would be blocked, then switch to ENFORCE in production:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# Shadow mode — see what governance would deny without blocking anything
governance = TealTigerMiddleware(
    mode=GovernanceMode.MONITOR,
    policies=[
        GovernancePolicy.tool_allowlist(["search", "read_*"]),
        GovernancePolicy.pii_block(["ssn", "credit_card"]),
    ],
)

# After validation, promote to ENFORCE
governance.mode = GovernanceMode.ENFORCE

Policy Types#

Tool Allowlist#

Restrict which tools the agent can call. Patterns are matched with fnmatch, so the full glob syntax (*, ?, [seq]) applies:

1

GovernancePolicy.tool_allowlist(["search", "read_*", "github_*"])

Any tool not matching the patterns is denied with reason code TOOL_NOT_ALLOWED and risk score 80.

PII Detection#

Block tool calls containing sensitive data in their arguments:

1

GovernancePolicy.pii_block(["ssn", "credit_card", "email", "phone"])

Passing no argument applies all four categories:

Denied with risk score 90. One reason code is appended per matched category, in the form PII_DETECTED:ssn, PII_DETECTED:email, and so on.

Secret Detection#

Block tool calls containing API keys, tokens, or credentials:

1

GovernancePolicy.secret_detection()

Denied with reason code SECRET_DETECTED and risk score 95 — the highest of the policy checks.

Cost Limit#

Enforce per-session cost ceilings:

1

GovernancePolicy.cost_limit(max_per_session=5.0)

Cumulative cost is tracked across all tool calls using cost_per_call (configurable, defaults to 0.002). Once the limit is reached, further calls are denied with BUDGET_EXCEEDED and risk score 70.

Factory-Level Budget#

In addition to the policy-level cost_limit, you can set a factory-level budget that is always enforced, with no policy needed:

1
2
3
4
5
6
7

governance = TealTigerMiddleware(
    budget_limit=10.0,      # Hard ceiling — always enforced
    cost_per_call=0.002,    # Estimated cost per tool call
    policies=[
        GovernancePolicy.tool_allowlist(["search", "read_*"]),
    ],
)

budget_limit defaults to infinity. It is checked before any policy runs, providing a spending cap that no policy ordering can bypass.

Kill Switch#

Freeze an agent immediately — this blocks its turns and all of its tool calls, regardless of policies:

1
2
3

governance.freeze("assistant")      # Block everything for this agent
governance.is_frozen("assistant")   # True
governance.unfreeze("assistant")    # Restore normal governance

Freezing matches on the exact agent name. There is no wildcard — to freeze several agents, freeze each by name:

1
2

for name in ("researcher", "analyst", "writer"):
    governance.freeze(name)

The kill switch is enforced at both the turn level (on_turn) and the tool level (on_tool_execution):

• ENFORCE mode: returns ToolErrorEvent, blocking the entire turn or tool call
• MONITOR mode: records a DENY decision but allows the action through
• OBSERVE mode: not evaluated (passthrough)

Kill switch decisions carry reason code AGENT_FROZEN and risk score 100 (maximum).

Audit Trail#

Every governance decision produces a structured GovernanceDecision:

1
2
3
4
5
6
7
8
9

for decision in governance.decisions:
    print(
        f"[{decision.action}] tool={decision.tool_name} "
        f"agent={decision.agent_name} "
        f"reason_codes={decision.reason_codes} "
        f"risk={decision.risk_score} "
        f"latency={decision.evaluation_time_ms:.2f}ms "
        f"cost=${decision.cumulative_cost:.4f}"
    )

Decision Contract Fields#

TEEC Receipts#

Every tool evaluation — executed or blocked — produces a TEECReceipt:

1
2
3
4
5

for receipt in governance.receipts:
    print(
        f"{receipt.execution_outcome} | {receipt.tool_name} | "
        f"agent={receipt.agent_name} | decision={receipt.decision_id}"
    )

Factory State API#

The factory exposes the accumulated state directly:

Callbacks#

React to governance decisions in real time:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

def on_deny(decision):
    if decision.action == "DENY":
        alert_ops_team(decision.tool_name, decision.reason_codes)

def on_receipt(receipt):
    audit_log.append(receipt)

governance = TealTigerMiddleware(
    policies=[GovernancePolicy.tool_allowlist(["search"])],
    on_decision=on_deny,       # Called for every decision (ALLOW and DENY)
    on_receipt=on_receipt,     # Called for every TEEC receipt
)

Use Across Multiple Agents#

One governance instance can be attached to several agents — decisions, receipts, cost, and frozen agents are shared across all of them. Here the coordinator delegates to two subagents via Agent.as_tool(), and all three run under the same policy set:

flowchart TB
    GOV["🛡️ TealTigerMiddleware — Shared Instance<br/>(decisions, receipts, cost, frozen_agents)"]
    GOV -->|evaluates| R["🔬 Researcher<br/>tools: search, read_*"]
    GOV -->|evaluates| AN["📊 Analyst<br/>tools: calculator"]
    GOV -->|"🧊 FROZEN"| W["✍️ Coordinator"]

    style GOV fill:#f0fdfa,stroke:#14b8a6,stroke-width:2px
    style R fill:#eff6ff,stroke:#93c5fd
    style AN fill:#eff6ff,stroke:#93c5fd
    style W fill:#fef2f2,stroke:#fca5a5

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

import asyncio

from ag2 import Agent
from ag2.config import AnthropicConfig
from ag2.extensions.tealtiger import GovernanceMode, GovernancePolicy, TealTigerMiddleware

# Single governance instance shared across all agents
governance = TealTigerMiddleware(
    mode=GovernanceMode.ENFORCE,
    policies=[
        GovernancePolicy.tool_allowlist(["search", "read_*", "calculator"]),
        GovernancePolicy.pii_block(["ssn", "credit_card", "email"]),
        GovernancePolicy.secret_detection(),
        GovernancePolicy.cost_limit(max_per_session=10.0),
    ],
)
config = AnthropicConfig(model="claude-sonnet-4-6")

async def main() -> None:
    researcher = Agent("researcher", config=config, middleware=[governance])
    analyst = Agent("analyst", config=config, middleware=[governance])
    coordinator = Agent(
        "coordinator",
        config=config,
        middleware=[governance],
        tools=[
            researcher.as_tool(description="Research a topic and return findings."),
            analyst.as_tool(description="Analyze findings and return a summary."),
        ],
    )
    reply = await coordinator.ask("Research AI governance trends, then analyze them.")
    print(await reply.content())
    # Cost, decisions, and receipts are shared across all three agents
    print(f"Total decisions: {len(governance.decisions)}")
    print(f"Denied: {governance.deny_count}")
    print(f"Total cost: ${governance.total_cost:.4f}")
    # Freeze one agent — it can no longer take turns or call tools
    governance.freeze("researcher")

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

Complete Example: Governed Research Agent#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

import asyncio

from ag2 import Agent
from ag2.config import AnthropicConfig
from ag2.extensions.tealtiger import GovernanceMode, GovernancePolicy, TealTigerMiddleware

governance = TealTigerMiddleware(
    mode=GovernanceMode.ENFORCE,
    budget_limit=10.0,
    cost_per_call=0.003,
    policies=[
        # Tool governance — only safe tools
        GovernancePolicy.tool_allowlist(["web_search", "calculator", "read_file"]),
        # Data protection — block PII leakage
        GovernancePolicy.pii_block(["ssn", "credit_card", "email", "phone"]),
        # Credential protection — block secret leakage
        GovernancePolicy.secret_detection(),
        # Cost governance — hard per-session limit
        GovernancePolicy.cost_limit(max_per_session=5.0),
    ],
    on_decision=lambda d: print(f"  [{d.action}] {d.tool_name} — {d.reason_codes}"),
)

async def main() -> None:
    agent = Agent(
        "research_assistant",
        config=AnthropicConfig(model="claude-sonnet-4-6"),
        middleware=[governance],
    )
    reply = await agent.ask("Research AI governance frameworks for enterprise deployment.")
    print(await reply.content())
    # Post-run analysis
    allowed = [d for d in governance.decisions if d.action == "ALLOW"]
    denied = [d for d in governance.decisions if d.action == "DENY"]
    print("\n--- Governance Summary ---")
    print(f"Total evaluations: {len(governance.decisions)}")
    print(f"Allowed: {len(allowed)}")
    print(f"Denied: {len(denied)}")
    print(f"Session cost: ${governance.total_cost:.4f}")
    print(f"TEEC receipts: {len(governance.receipts)}")
    for d in denied:
        print(f"  • {d.tool_name}: {d.reason_codes} (risk={d.risk_score})")

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

Evaluation Order#

Two checks always run first, in this order. Everything after them is your policy list, walked in the order you declared it — the first DENY wins and short-circuits the rest:

flowchart TD
    START([Tool Call Received]) --> S1
    S1{{"1️⃣ Kill Switch<br/>(risk: 100)"}} -->|frozen| DENY[❌ DENY]
    S1 -->|not frozen| S2
    S2{{"2️⃣ Factory Budget<br/>(risk: 70)"}} -->|exceeded| DENY
    S2 -->|within budget| S3
    S3{{"3️⃣ Your policies, in list order<br/>first DENY wins"}} -->|violation| DENY
    S3 -->|all clean| ALLOW[✅ ALLOW]

    style S1 fill:#fef2f2,stroke:#fca5a5
    style S2 fill:#fef9c3,stroke:#fde047
    style S3 fill:#eff6ff,stroke:#93c5fd
    style DENY fill:#fef2f2,stroke:#dc2626,stroke-width:2px
    style ALLOW fill:#f0fdf4,stroke:#16a34a,stroke-width:2px

Each policy type carries its own risk score when it denies:

If all checks pass, the decision is ALLOW with reason code POLICY_ALLOW and risk score 0.

Architecture#

flowchart TB
    subgraph Factory["TealTigerMiddleware (Long-lived factory)"]
        STATE["_decisions | _receipts<br/>_frozen_agents | _cumulative_cost"]
        API["freeze() | unfreeze() | is_frozen()<br/>.decisions | .receipts | .total_cost"]
    end

    Factory -->|"creates per turn"| PerTurn

    subgraph PerTurn["_TealTigerPerTurn (Per-turn instance)"]
        TURN["on_turn()<br/>Kill switch + mode gating"]
        TOOL["on_tool_execution()<br/>Policy evaluation + receipts"]
    end

    PerTurn -->|"writes back"| STATE

    style Factory fill:#f0fdfa,stroke:#14b8a6,stroke-width:2px
    style PerTurn fill:#eff6ff,stroke:#3b82f6,stroke-width:2px

ag2/extensions/tealtiger/
├── __init__.py       # Public API: TealTigerMiddleware, GovernanceMode, GovernancePolicy, GovernanceDecision, TEECReceipt
├── types.py          # GovernanceMode, GovernancePolicy, GovernanceDecision, TEECReceipt
└── middleware.py     # TealTigerMiddleware (factory) + _TealTigerPerTurn (per-turn) + _PII_PATTERNS / _SECRET_PATTERNS

The middleware follows AG2's middleware factory pattern:

• TealTigerMiddleware is the long-lived factory holding shared state (decisions, receipts, frozen agents, cumulative cost)
• Calling the factory creates a _TealTigerPerTurn instance for the turn, holding a reference back to the factory
• on_turn handles kill switch enforcement at the turn level
• on_tool_execution handles all policy evaluation at the tool level
• Kill switch, budget, and audit state persist across turns and agents

Performance#

Governance runs entirely in-process: no network calls, no LLM inference, and no external service dependencies. Every check is a fnmatch glob or a precompiled re pattern over the serialized tool arguments, so evaluation cost scales with argument size rather than with model latency.

Each decision records its own measured cost in evaluation_time_ms, so you can profile governance overhead against your own workload:

1
2

slowest = max(governance.decisions, key=lambda d: d.evaluation_time_ms)
print(f"{slowest.tool_name}: {slowest.evaluation_time_ms:.3f}ms")

Links#

• TealTiger AG2 integration guide — advanced capabilities beyond the bundled extension
• TealTiger on GitHub (Apache 2.0)
• TealTiger governance platform on PyPI

Maintainer: @nagasatish007 / TealTiger.