TealTiger

The ag2.extensions.tealtiger module adds deterministic governance guardrails to AG2 agents. TealTigerMiddleware enforces tool allowlists and blocklists, validates tool arguments, detects PII and secrets in tool arguments and in tool results, blocks prompt injection attempts, 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
8

from ag2.extensions.tealtiger import (
    GovernanceDecision,
    GovernanceMode,
    GovernancePolicy,
    OutputAction,
    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
30

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.prompt_injection_block(),
        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-5"),
        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 three 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/>pre-tool}
    E -->|"✅ ALLOW"| F[⚡ Execute Tool]
    E -->|"❌ DENY"| G[🚫 ToolErrorEvent]
    F --> I{🛡️ Stage 3<br/>output_scan}
    I -->|"✅ clean or redacted"| H[📝 TEEC Receipt]
    I -->|"❌ BLOCK"| G
    G --> H

    style B fill:#f0fdfa,stroke:#14b8a6,stroke-width:2px
    style E fill:#f0fdfa,stroke:#14b8a6,stroke-width:2px
    style I 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

All three 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.

Tool Blocklist#

The complement of the allowlist: allow every tool except the ones you name. Reach for this when an agent has many safe tools and only a few dangerous ones, so enumerating every safe tool would be tedious. Patterns are matched with fnmatch, so the full glob syntax (*, ?, [seq]) applies:

1

GovernancePolicy.tool_blocklist(["delete_*", "shell", "drop_table"])

Any tool matching a pattern is denied with reason code TOOL_BLOCKED and risk score 80.

An empty blocklist raises ValueError at policy construction, so a policy that blocks nothing cannot be created by mistake.

You can combine an allowlist with a blocklist — for example, permit everything under read_* but still deny read_secrets. When both are configured, every policy is evaluated until one denies, so either one denying results in a DENY; the policy that denies is the one reported, whatever its position in the list:

1
2
3
4
5
6
7
8

from ag2.extensions.tealtiger import GovernancePolicy, TealTigerMiddleware

governance = TealTigerMiddleware(
    policies=[
        GovernancePolicy.tool_allowlist(["read_*"]),
        GovernancePolicy.tool_blocklist(["read_secrets"]),
    ],
)

Here read_file passes both policies, while read_secrets clears the allowlist and is denied by the blocklist — so the reason code is TOOL_BLOCKED, from the second entry.

Argument Validation#

Where the allowlist and blocklist govern which tools run, arg_validation governs what a tool is called with — a defense against dangerous argument values such as SQL injection, path traversal, or oversized payloads. It applies to any tool whose name matches tool via fnmatch, and checks the named arguments against per-argument constraints:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# Reject long queries and dangerous SQL verbs
GovernancePolicy.arg_validation(
    "sql_query",
    {"query": {"max_length": 500, "blocked_terms": ["DROP", "DELETE", ";--"]}},
)

# Reject path traversal in a file tool
GovernancePolicy.arg_validation(
    "read_file",
    {"path": {"blocked_patterns": [r"\.\.[\\/]"]}},  # ../ and ..\
)

Each argument's constraint spec may combine any of these checks:

A violation is denied with reason code ARG_VALIDATION:{argument}:{check} (for example ARG_VALIDATION:query:max_length) and risk score 85. Only the arguments named in constraints are checked — any others pass through — and a constrained argument that is absent from the call is skipped.

The policy validates its own spec at construction, checking each check's value and not just its name. An empty tool or constraints, a non-dict spec, an unknown check name, an unsupported type, a negative or non-integer length bound, a max_length below the min_length, an empty term or value list, or an uncompilable regex all raise ValueError. A typo therefore surfaces where you wrote it — rather than leaving a policy that silently validates nothing, or one that raises from inside the governance path on the first call it evaluates.

Patterns are compiled during that same validation, so evaluation never pays for compilation:

1
2

GovernancePolicy.arg_validation("read_file", {"path": {"blocked_patterns": ["[unclosed"]}})
# ValueError: Invalid regex in `blocked_patterns` for argument 'path': '[unclosed' — ...

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.

Output Scanning#

Every policy above inspects a tool's arguments before it runs. output_scan inspects what a tool returns — the data-leakage direction. A tool that reads a database, a file, or an external API can hand back an SSN or a leaked credential that would otherwise land straight in the model's context on the next turn. This scans the result first and redacts, blocks, or flags it.

1
2
3
4
5
6

# Redact PII, block secrets (the defaults)
GovernancePolicy.output_scan()

# Block on any PII too, and only scan the categories you care about.
# Actions take an OutputAction member or its string name, as `mode` takes GovernanceMode.
GovernancePolicy.output_scan(pii_action=OutputAction.BLOCK, categories=["ssn", "credit_card"])

Each detector carries its own action:

The defaults mirror the sensible split: PII is redacted (the agent usually still needs the surrounding result), secrets are blocked (a leaked credential should not reach the model at all).

The two detectors act independently: output_scan(pii_action="FLAG", secret_action="REDACT") records the PII it finds and rewrites only the credential. Where they overlap — a single result carrying both, or several output_scan policies disagreeing about the same detector — the more restrictive action wins.

Results are scanned whatever shape they arrive in. A tool returning a dict or a list leaks just as readily as one returning a sentence, so structured results are scanned and redacted through their strings:

1
2
3
4
5
6

# A tool whose result is a dict, not a sentence
def lookup_customer(name: str) -> dict[str, str]:
    return {"name": name, "ssn": "123-45-6789"}

# With GovernancePolicy.output_scan() configured, the model receives:
#   {"name": "Ada", "ssn": "[REDACTED:ssn]"}

Findings are recorded with reason codes OUTPUT_PII_DETECTED:{category} and OUTPUT_SECRET_DETECTED, plus OUTPUT_REDACTED, OUTPUT_REDACTION_INCOMPLETE, or OUTPUT_BLOCKED describing what was done. Risk score is 90 when a secret is present, 60 for PII only.

Error results are scanned too. A tool that raises with an SSN or credential in its exception message leaks it into the model's context just as a returned value would, so output_scan scans a ToolErrorEvent's message and traceback on the same terms. An error stays an error: its content is redacted in place, and a BLOCK in ENFORCE replaces it with a sanitized governance error rather than turning the failure into a success.

output_scan validates its own configuration: scanning nothing (scan_pii=False, scan_secrets=False), an invalid action, an unknown PII category, or scan_pii=True with an empty categories list all raise ValueError at construction, so a policy that would silently scan nothing cannot be created by mistake.

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.

Prompt Injection Detection#

Block tool calls containing adversarial prompt injection patterns in their arguments:

1

GovernancePolicy.prompt_injection_block()

Detects 5 technique categories with 20 regex patterns:

Denied with reason code PROMPT_INJECTION:{technique}/{pattern_name} and risk score 95.

Patterns match on injection framing, not on keywords alone — DAN counts only alongside a jailbreak cue and is matched case-sensitively, a <system> tag counts only when instruction-like content follows it. Without that, ordinary arguments (a colleague named Dan, <system> in an XML payload, "add the new rules to the linter config") would be denied at risk score 95.

Selective techniques — enable only specific categories:

1
2
3

GovernancePolicy.prompt_injection_block(
    techniques=["instruction_override", "role_manipulation"],
)

Confidence threshold — tune sensitivity to reduce false positives:

1
2
3

GovernancePolicy.prompt_injection_block(
    confidence_threshold=0.85,  # Skip lower-confidence patterns
)

Each pattern carries a fixed confidence score (0.0–1.0). The threshold selects which patterns run — it is not a score computed per match — and only patterns at or above it are evaluated. Default threshold is 0.7, which leaves rot13_reference (0.65) inactive; lower the threshold to enable it.

An unknown technique name, an empty techniques list, or a threshold outside 0.0–1.0 raises ValueError at policy construction, so a typo cannot silently leave the policy detecting nothing.

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-5")

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-5"),
        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. Once the tool has run, output_scan gets the last word on its result:

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| RUN[/Tool executes/]
    RUN --> S4
    S4{{"4️⃣ Output scan, on the result<br/>(risk: 90 secret / 60 PII)"}} -->|BLOCK| DENY
    S4 -->|REDACT| SANITIZED[✅ ALLOW<br/>sanitized result]
    S4 -->|FLAG or clean| ALLOW[✅ ALLOW]

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

Steps 1–3 govern whether the tool runs; step 4 governs what its result is allowed to carry back. Only step 4 runs after execution, and it is skipped entirely in OBSERVE mode.

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 + injection detection types
├── types.py          # GovernanceMode, GovernancePolicy, GovernanceDecision, TEECReceipt + injection detection types (InjectionPattern, InjectionFinding, INJECTION_PATTERNS, INJECTION_TECHNIQUES, DEFAULT_INJECTION_CONFIDENCE_THRESHOLD)
└── middleware.py     # TealTigerMiddleware (factory) + _TealTigerPerTurn (per-turn) + _PII_PATTERNS / _SECRET_PATTERNS / _detect_prompt_injection

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, a precompiled re pattern, or a plain comparison over the tool arguments — every regex the middleware evaluates, arg_validation's blocked_patterns included, is compiled before the first call reaches it. Evaluation cost therefore 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.