AG2 Integration with TealTiger

TealTiger ships as a built-in AG2 extension: deterministic governance guardrails for AG2 agents with no LLM in the governance path. It enforces tool allowlists and blocklists, detects PII and secrets in tool arguments, blocks prompt injection, tracks per-session cost, provides a per-agent kill switch, and emits structured TEEC audit receipts.

Package: Ships with AG2 (built-in extension — no separate install) License: Apache 2.0 Latency: Under 2ms per evaluation Dependencies: Zero — pure AG2 + stdlib Source: ag2/extensions/tealtiger/ PR: #3128 (merged)

This page documents the built-in AG2 extension, which is intentionally focused and dependency-free. The standalone TealTiger platform adds capabilities that are not part of the bundled extension — rate limiting, circuit breakers, content moderation, evidence export, and OpenTelemetry spans. Those are called out in Beyond the built-in extension.

Installation

TealTiger ships as a built-in AG2 extension — no separate package needed:

pip install ag2

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

Quick Start

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

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

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

agent = Agent(
    "assistant",
    config=AnthropicConfig(model="claude-sonnet-4-6"),
    middleware=[governance],
)

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

Multi-Stage Defense

TealTiger governs at two stages of the AG2 agent lifecycle:

User Input → [Stage 1: on_turn — kill switch] → Agent → Tool Call → [Stage 2: on_tool_execution — policy evaluation] → Execute or Deny

Hook	Stage	What It Does
`on_turn`	Turn defense	Kill switch enforcement — a frozen agent cannot take a turn
`on_tool_execution`	Pre-tool defense	Tool allowlist/blocklist, prompt injection, PII scan, secret scan, cost limit

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

A TEECReceipt is emitted for every tool evaluation, whether the call was executed or blocked. Turn-level kill switch denials record a GovernanceDecision but do not emit a receipt.

Governance Modes

Mode	Behavior	Use Case
ENFORCE	Evaluates all policies, blocks violations with a `ToolErrorEvent`	Production
MONITOR	Evaluates all policies, records decisions, never blocks	Staging / shadow testing
OBSERVE	Skips policy evaluation, tracks cost, records an `OBSERVE_PASSTHROUGH` decision	Initial rollout / lowest overhead

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

governance = TealTigerMiddleware(
    mode=GovernanceMode.MONITOR,
    policies=[GovernancePolicy.tool_allowlist(["search", "read_*"])],
)

# After validation
governance.mode = GovernanceMode.ENFORCE

Policy Types

All policies are created with GovernancePolicy factory methods and evaluated deterministically over the serialized tool arguments.

Tool Allowlist

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

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

A tool not matching any pattern 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 most tools are safe and you only need to carve out a dangerous few. Patterns also match with fnmatch:

GovernancePolicy.tool_blocklist(["delete_*", "execute_sql", "send_email"])

A tool matching any pattern is denied with reason code TOOL_BLOCKED and risk score 80. An empty blocklist raises ValueError at construction, so a policy that blocks nothing can’t be created by mistake. Allowlist and blocklist compose — allow read_* but still deny read_secrets. When both are configured, either policy denying the call results in a DENY (the first match in your list is the one reported):

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

PII Detection

Block tool calls containing PII in their arguments:

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

Passing no argument applies all four categories:

Category	Example Match
`ssn`	123-45-6789
`credit_card`	4111-1111-1111-1111
`email`	[email protected]
`phone`	+1-555-123-4567

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 in their arguments:

GovernancePolicy.secret_detection()

Pattern	What It Catches
`sk-…` (20+ chars)	OpenAI-style API keys
`ghp_…` (36+ chars)	GitHub personal access tokens
`AKIA…` (16 chars)	AWS access key IDs
`xox[bpors]-…`	Slack tokens
`api_key=…` / `apikey:…` (20+ chars)	Generic API keys
PEM private key headers	RSA, EC, DSA private keys

Denied with reason code SECRET_DETECTED and risk score 95.

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 return an SSN or a leaked credential that would otherwise land in the model’s context on the next turn. This scans the result and redacts, blocks, or flags it first.

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

# Block on any PII too, and scan only the categories you name
GovernancePolicy.output_scan(pii_action="BLOCK", categories=["ssn", "credit_card"])

Each detector carries its own action:

Action	Effect
`REDACT`	Replaces matched values with `[REDACTED:{type}]`; the sanitized result flows through.
`BLOCK`	Withholds the whole result and fails the turn with a governance error (ENFORCE only). In MONITOR it degrades to `REDACT` so the value still never leaks.
`FLAG`	Records the finding in the decision and receipt trail; passes the result through unchanged.

Defaults follow the sensible split: PII is redacted (you usually still need the surrounding result), secrets are blocked (a leaked credential should not reach the model at all). When both a PII and a secret finding occur, the more restrictive action wins. Findings are recorded with reason codes OUTPUT_PII_DETECTED:{category} and OUTPUT_SECRET_DETECTED, plus OUTPUT_REDACTED or OUTPUT_BLOCKED describing the action. Risk score is 90 when a secret is present, 60 for PII only. Output scanning runs after the tool executes (it governs the result), in MONITOR and ENFORCE; in OBSERVE the result passes through unmodified.

Prompt Injection Detection

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

GovernancePolicy.prompt_injection_block()

Detects 5 technique categories with 20 regex patterns:

Technique	Patterns	What It Catches
`instruction_override`	5	”Ignore previous instructions”, system prompt override, context resets
`role_manipulation`	5	DAN jailbreak, developer mode, persona switching
`context_manipulation`	4	Delimiter injection, XML/chat-template tag injection, fake system messages
`encoding_evasion`	4	Base64 payloads, hex/unicode escapes, ROT13 references
`multi_turn_assembly`	2	Payload splitting, continuation attacks

Denied with reason code PROMPT_INJECTION:{technique}/{pattern_name} and risk score 95. Enable only specific techniques, or tune the confidence threshold to reduce false positives:

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

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 is 0.7. An unknown technique name, an empty techniques list, or a threshold outside 0.0–1.0 raises ValueError at construction.

Policies are evaluated in the order you pass them, and evaluation stops at the first denial. Put prompt_injection_block() before tool_allowlist(...) if you want injection attempts reported as PROMPT_INJECTION:… rather than TOOL_NOT_ALLOWED.

Cost Limit

Enforce a per-session cost ceiling:

GovernancePolicy.cost_limit(max_per_session=5.0)

Cumulative cost is tracked across all tool calls using cost_per_call (a middleware option, default 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, set a factory-level budget that is always enforced before any policy runs — a spending cap no policy ordering can bypass:

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.

Kill Switch

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

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:

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: returns a ToolErrorEvent, blocking the turn or tool call
MONITOR: records a DENY decision but allows the action through
OBSERVE: not evaluated (passthrough)

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

Audit Trail

Every governance decision produces a structured GovernanceDecision:

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

Field	Type	Description
`decision_id`	`str`	UUID4 string, unique per decision
`action`	`str`	`ALLOW` or `DENY`
`mode`	`str`	`OBSERVE`, `MONITOR`, or `ENFORCE`
`agent_name`	`str`	Agent that made the call, or `unknown`
`tool_name`	`str`	Tool evaluated (`*` for turn-level denials)
`reason_codes`	`list[str]`	`TOOL_NOT_ALLOWED`, `TOOL_BLOCKED`, `PROMPT_INJECTION:{technique}/{pattern}`, `PII_DETECTED:{category}`, `SECRET_DETECTED`, `OUTPUT_PII_DETECTED:{category}`, `OUTPUT_SECRET_DETECTED`, `OUTPUT_REDACTED`, `OUTPUT_BLOCKED`, `BUDGET_EXCEEDED`, `AGENT_FROZEN`, `POLICY_ALLOW`, `OBSERVE_PASSTHROUGH`
`risk_score`	`int`	0 (safe) to 100 (critical)
`evaluation_time_ms`	`float`	Governance latency for this decision
`cost_tracked`	`float`	Cost attributed to this call in USD
`cumulative_cost`	`float`	Running session cost in USD
`timestamp_ms`	`float`	Unix timestamp in milliseconds

TEEC Receipts

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

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

Receipt fields: receipt_id, decision_id, agent_name, tool_name, action, execution_outcome (executed, blocked, or error), reason_codes, risk_score, policy_digest (short SHA-256 of the active policy set), and timestamp_ms.

Factory State API

The factory exposes accumulated state directly:

Member	Description
`decisions`	All `GovernanceDecision` records across all turns
`receipts`	All `TEECReceipt` records across all turns
`total_cost`	Cumulative tracked cost in USD
`deny_count`	Number of decisions with `action == "DENY"`
`is_frozen(name)`	Whether that agent is currently frozen
`reset()`	Clear decisions, receipts, cost, and frozen agents

Callbacks

React to governance decisions in real time:

governance = TealTigerMiddleware(
    policies=[GovernancePolicy.tool_allowlist(["search"])],
    on_decision=lambda d: print(f"[{d.action}] {d.tool_name} {d.reason_codes}"),
    on_receipt=lambda r: audit_log.append(r),
)

on_decision is called for every decision (ALLOW and DENY); on_receipt is called for every TEEC receipt.

Evaluation Order

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

Kill switch — AGENT_FROZEN, risk 100
Factory budget — BUDGET_EXCEEDED, risk 70
Your policies, in list order — first DENY wins

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

Policy	Reason Code	Risk Score
Kill switch	`AGENT_FROZEN`	100
Prompt injection	`PROMPT_INJECTION:{technique}/{pattern}`	95
Secret detection	`SECRET_DETECTED`	95
PII detection	`PII_DETECTED:{category}`	90
Tool allowlist	`TOOL_NOT_ALLOWED`	80
Tool blocklist	`TOOL_BLOCKED`	80
Cost limit / factory budget	`BUDGET_EXCEEDED`	70
Output scan (post-tool, on the result)	`OUTPUT_SECRET_DETECTED` / `OUTPUT_PII_DETECTED:{category}`	90 / 60

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

Output scanning is the one check that runs after the tool executes rather than before it — it governs what the result puts back into the model’s context, so it cannot short-circuit the pre-tool order above.

Because policies are evaluated in the order you pass them, the reason code on a denial depends on that order. If a call violates two policies, only the first in your list is reported. Put the checks you most want attributed earliest.

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:

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

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

researcher = Agent("researcher", config=config, middleware=[governance])
analyst = Agent("analyst", config=config, middleware=[governance])

# Freeze one agent — it can no longer take turns or call tools
governance.freeze("researcher")

print(f"Total decisions: {len(governance.decisions)}")
print(f"Denied: {governance.deny_count}")
print(f"Total cost: ${governance.total_cost:.4f}")

Complete Example: Governed Research Agent

import asyncio

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

governance = TealTigerMiddleware(
    mode=GovernanceMode.ENFORCE,
    budget_limit=10.0,
    cost_per_call=0.003,
    policies=[
        GovernancePolicy.prompt_injection_block(),
        GovernancePolicy.tool_allowlist(["web_search", "calculator", "read_file"]),
        GovernancePolicy.tool_blocklist(["delete_*"]),
        GovernancePolicy.pii_block(["ssn", "credit_card", "email", "phone"]),
        GovernancePolicy.secret_detection(),
        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())

    denied = [d for d in governance.decisions if d.action == "DENY"]
    print("\n--- Governance Summary ---")
    print(f"Total evaluations: {len(governance.decisions)}")
    print(f"Denied: {len(denied)}")
    print(f"Session cost: ${governance.total_cost:.4f}")
    print(f"TEEC receipts: {len(governance.receipts)}")

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

Beyond the built-in extension

The built-in extension is deliberately small and dependency-free. The standalone TealTiger platform adds governance capabilities that are not part of the bundled AG2 extension:

Capability	Built-in extension	Standalone platform
Tool allowlist / blocklist	✅	✅
PII detection (tool args)	✅	✅
Secret detection (tool args)	✅	✅
Prompt injection blocking	✅	✅
Cost limit + factory budget	✅	✅
Per-agent kill switch	✅	✅
TEEC receipts + decision trail	✅	✅
Output scanning (PII redaction / secret blocking on tool results)	✅	✅
Rate limiting	—	✅
Circuit breaker	—	✅
Content moderation	—	✅
Per-tool argument constraints	—	✅
Evidence export (SARIF / JUnit / JSON)	—	✅
OpenTelemetry spans	—	✅

If you need those, see the TealTiger platform documentation. The built-in extension and the platform share the same policy vocabulary, so configurations translate directly.

OWASP AI Security Coverage

The built-in extension addresses several OWASP Agentic Security Issues:

OWASP ASI	Covered	How
ASI-01: Excessive Agency	✅	Tool allowlist/blocklist, kill switch
ASI-03: Prompt Injection	✅	Deterministic injection pattern detection
ASI-04: Sensitive Information Disclosure	✅	PII detection in tool arguments
ASI-08: Excessive Resource Consumption	✅	Cost limit + factory budget
ASI-09: Insufficient Monitoring	✅	Decision trail + TEEC receipts

Broader coverage (output handling, sandboxing constraints, circuit breakers) is available through the standalone platform.

Key Properties

Property	Detail
Deterministic	No LLM in the governance path — same input, same decision
Fast	Under 2ms per evaluation
Two-stage	Turn-level kill switch + pre-tool policy evaluation
Auditable	Decision trail and TEEC receipts, shared across agents
Multi-agent	One instance governs many agents with shared state
Zero external deps	All evaluation in-process, no API keys needed
Built-in	Ships with AG2 — no separate package or install

Architecture

The middleware follows AG2’s middleware factory pattern:

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

TealTigerMiddleware — long-lived factory holding shared state (decisions, receipts, frozen agents, cumulative cost)
Calling the factory creates a _TealTigerPerTurn instance for the turn, referencing the factory
on_turn handles kill switch enforcement; on_tool_execution handles policy evaluation
Kill switch, budget, and audit state persist across turns and agents

Troubleshooting

Policies not enforcing

Ensure mode=GovernanceMode.ENFORCE. In MONITOR or OBSERVE mode, violations are recorded but not blocked. Check governance.decisions to confirm policies evaluated.

Kill switch not working

Verify the agent name matches exactly: governance.freeze("assistant") must match the name passed to Agent("assistant", ...). The freeze is case-sensitive with no wildcard.

Cost tracking shows zero

Set cost_per_call on the middleware. Without it, the tracker cannot estimate spend (it defaults to 0.002 per call).

Looking for rate limiting, circuit breakers, or evidence export

Those are part of the standalone TealTiger platform, not the built-in extension. See Beyond the built-in extension.

​AG2 Integration with TealTiger

​Installation

​Quick Start

​Multi-Stage Defense

​Governance Modes

​Policy Types

​Tool Allowlist

​Tool Blocklist

​PII Detection

​Secret Detection

​Output scanning

​Prompt Injection Detection

​Cost Limit

​Factory-Level Budget

​Kill Switch

​Audit Trail

​Decision Contract Fields

​TEEC Receipts

​Factory State API

​Callbacks

​Evaluation Order

​Use Across Multiple Agents

​Complete Example: Governed Research Agent

​Beyond the built-in extension

​OWASP AI Security Coverage

​Key Properties

​Architecture

​Troubleshooting

​Links

AG2 Integration with TealTiger

Installation

Quick Start

Multi-Stage Defense

Governance Modes

Policy Types

Tool Allowlist

Tool Blocklist

PII Detection

Secret Detection

Output scanning

Prompt Injection Detection

Cost Limit

Factory-Level Budget

Kill Switch

Audit Trail

Decision Contract Fields

TEEC Receipts

Factory State API

Callbacks

Evaluation Order

Use Across Multiple Agents

Complete Example: Governed Research Agent

Beyond the built-in extension

OWASP AI Security Coverage

Key Properties

Architecture

Troubleshooting

Links