rgltr: Tool Governance for Pydantic AI Agents

rgltr (regulator) controls which tools a Pydantic AI agent can use and when. It provides decorators for specifying what tools are provided to the LLM based on the data exposed to the current context window and authorization policies for verifying whether the attempted tool call is allowed under the current conditions.

rgltr prevents data exfiltration via tool calls by disallowing dangerous tools (defined by you) to be called when sensitive data (also defined by you) has been provided to the LLM. (The logic behind this is that once the LLM has access to tools it can pass any data in its context to any tool. If a tool sends data out from the system, the safe way is to provide this tool to the LLM only when it has access to data that can be sent outside.)

Installation

pip install rgltr pydantic-ai

Simple example of tagging data returned by a tool as sensitive and blocking LLM calls to another tool when this tag is active:

from enum import Enum
from pydantic_ai import Agent
from rgltr import tag
from rgltr.pydantic_ai import RegulatedAgent

class Tag(Enum):
    CUSTOMERS = "customers"

agent = Agent("openai:gpt-4")

# This tool returns sensitive information that should not be posted to Slack.
# When the tool is called the CUSTOMERS tag is activated. It stays active during the run context.
@agent.tool_plain
@tag(activates=[Tag.CUSTOMERS])
async def get_customer(customer_id: str) -> dict:
    return await db.get_customer(customer_id)

# This tool is blocked by CUSTOMERS tag. When the tag is active the LLM cannot call the tool.
@agent.tool_plain
@tag(blocked_by=[Tag.CUSTOMERS])
async def post_to_slack(message: str) -> None:
    await slack.post(message)

agent = RegulatedAgent(agent, agent_id="customer-support")

result = await agent.run("Look up customer 123 and post to Slack")

(The above example, and the following ones, target function tools. MCP tools use the same decorators and rules; see the Complete Example.)

1. Run starts, no tags active → Tools visible: get_customer, post_to_slack

2. LLM calls get_customer("123") → Executes, returns customer data → CUSTOMERS tag activated

3. rgltr rebuilds tool list for LLM → get_customer: no blocked_by → visible → post_to_slack: CUSTOMERS active → HIDDEN

4. LLM tries to call post_to_slack (hallucinated or from memory of earlier turns) → Tool not in list of allowed tools → Pydantic AI returns generic error: "Unknown tool 'post_to_slack'" → Error includes available tools, LLM can recover

Note: rgltr decorators (@tag and others) are declarations, not guarantees of what the tool actually does. You define the annotations and rgltr manages tools according to them.

See here for more usage examples

How Governance Works

rgltr controls tools (local function tools and MCP) through two mechanisms that operate at different times:

Tool Visibility (list of tools exposed to the LLM)

Before each message is sent to the LLM, rgltr builds the tool list:

1. Apply static filtering rules by name for tools (function and MCP, more on this later)
2. Check which tags are currently active in the context
3. Hide tools where:
   • Tool has blocked_by=[Tag.X] and Tag.X is active
   • Tool is assigned to a boundary configured to be hidden by active tags

Policy Enforcement

When the LLM calls a tool, rgltr runs policy checks:

1. Execute all policy functions attached to the tool
2. If any policy returns deny() → return error message to LLM
3. If any policy returns requires_approval() → raise exception for host app
4. If all policies return allow() → execute the tool

The Governance Flow

flowchart TD
  U[User message] --> B[Build tool list based on filters and tags]
  B --> L[Send list of visible tools to LLM]
  L --> C[LLM calls a tool]
  C --> P[Policy check]
  P -->|allow| X[Execute tool]
  P -->|deny| E[Return error to LLM]
  P -->|requires approval| A[Raise approval exception]
  X --> T[Tool activates tags]
  T --> B

Loading

What rgltr does not do

• Automatically infer what kind of data a tool returns (rgltr only controls annotated tools)
• Verify what a tool actually does (rgltr does not track data flow; it only enforces tool-level rules)
• Sanitize or redact data returned by tool calls (that's the tool's responsibility)

Core Concepts

rgltr uses the following decorators:

Concept	Purpose	Example
@tag(activates)	Activate tags when tool is called	@tag(activates=[Tag.CUSTOMERS])
@tag(blocked_by)	Hide tool when tag(s) are active	@tag(blocked_by=[Tag.INTERNAL])
@boundary	Assign tool to a boundary (global rules may hide it when tags are active)	@boundary(Boundary.EXTERNAL)
@policy	Runtime authorization when tool is called	@policy(require=[is_admin])
@governed	Mark tool as governed for audit report (without usage restrictions)	@governed()

Tags

Tags represent data categories. They're user-defined enums:

class Tag(Enum):
    CUSTOMERS = "customers"
    FINANCIAL = "financial"
    INTERNAL = "internal"

Tags are activated when tools execute, and they accumulate during a run. Once activated, a tag stays active until the context ends.

The @tag decorator controls tag activation and tool visibility:

activates - When this tool is called, the listed tags are activated (and get added to the context).

@tag(activates=[Tag.CUSTOMERS])
async def get_customer(customer_id: str) -> dict:
    """When called, adds CUSTOMERS to active tags."""
    return await db.get_customer(customer_id)

blocked_by - This tool becomes hidden from the LLM when any of the listed tags are active:

@tag(blocked_by=[Tag.CUSTOMERS])
async def export_to_csv(data: dict) -> str:
    """Hidden from LLM when CUSTOMERS tag is active."""
    return convert_to_csv(data)

A tool can both activate tags and be conditionally hidden:

@tag(activates=[Tag.INTERNAL], blocked_by=[Tag.EXTERNAL])
async def get_internal_config() -> dict:
    """Activates INTERNAL tag. Hidden if EXTERNAL is active."""
    return await db.get_config()

Note: A tool either activates a tag or it doesn't. There's no conditional activation based on arguments or results. If get_customer("wrong-id") returns nothing, the CUSTOMERS tag still activates. This is intentional: simpler to audit, harder to get wrong.

If you need different behaviors, split the logic into separate tools:

@tag(activates=[Tag.CUSTOMERS])
async def get_customer_details(id: str) -> dict:
    """Full customer record including PII."""
    return await db.get_customer(id)

# No tags defined: this tool is always visible to the LLM
async def customer_exists(id: str) -> bool:
    """Check if customer exists."""
    return await db.exists("customers", id)

Tag Context

Tags accumulate during a run. Use rgltr.context() to control scope:

import rgltr

# Each request gets isolated tag state
async def handle_request(request):
    async with rgltr.context():
        result = await agent.run("...", deps=request.user)
        # Tags automatically cleared when context exits
    return result

Manual tag management

async with rgltr.context() as ctx:
    ctx.activate([Tag.SENSITIVE])
    result = await agent.run("...", deps=request.user)

Multi-turn conversations

Tags from previous turns are automatically restored when message_history is provided:

# Turn 1
result1 = await agent.run("Look up customer 123")
# get_customer() called → CUSTOMERS tag activated

# Turn 2
result2 = await agent.run(
    "Post that to Slack",
    message_history=result1.all_messages(),
)
# rgltr scans history, sees get_customer() was called
# CUSTOMERS tag restored → post_to_slack hidden

rgltr scans the message history for tool call results. For each tool found, it looks up the tool's current activates configuration and re-activates those tags. This uses the current tool definitions—if you change a tool's activates between sessions, the new config applies.

Boundaries

Boundaries represent logical zones in your system architecture. They define which tags can reach a tool through centrally configured visibility rules.

Example:

class Boundary(Enum):
    EXTERNAL = "external"   # Third-party services (Slack, webhooks)
    USER = "user"           # Responses to the user
    INTERNAL = "internal"   # Internal systems (logging, metrics, databases)
    PARTNER = "partner"     # Trusted partner APIs

Tools marked with @boundary(Boundary.X) are subject to the visibility rules configured for that boundary. Configure which tags hide which boundaries to control what data categories can reach which tools.

@boundary(Boundary.EXTERNAL)
async def post_to_slack(message: str) -> None:
    """Sends data to external service."""
    await slack.post(message)

@boundary(Boundary.USER)
async def respond_to_user(message: str) -> str:
    """Returns data to the user."""
    return message

How Tags and Boundaries Connect

You must explicitly configure which tags hide which boundaries.

rgltr supports two configuration scopes:

Scope	API	Use case
Global	configure(...)	Shared defaults for all agents
Per-agent	RegulatedAgent(..., config=...)	Agent-specific overrides

Global configuration (affects all agents):

configure(
    boundaries={
        # All tags hide tools on EXTERNAL boundary
        Boundary.EXTERNAL: True,

        # Only specific tags hide tools on PARTNER boundary
        Boundary.PARTNER: [Tag.CUSTOMERS, Tag.FINANCIAL],

        # Only internal data hidden from USER
        Boundary.USER: [Tag.INTERNAL_ONLY],

        # INTERNAL: omitted = never hidden (has no effect on tool visibility)
    }
)

Agent-specific configuration overrides global:

agent = RegulatedAgent(
    Agent(...),
    config=RegulatorConfig(
        boundaries={},       # Nothing hidden
    ),
)

Composition rule: If a tool has both @boundary and @tag(blocked_by=...), restrictions are additive. The tool is hidden if any configured rule matches (global boundary rules OR local blocked_by).

The @governed Decorator

Use @governed() when you want a tool included in audit/coverage reports but don’t want to apply any restrictions. The tool functions just like a normal Pydantic AI tool.

Using Decorators with Function Tools

Decorate functions with rgltr decorators. If you’re using RegulatedAgent, it will wrap tools automatically; otherwise wrap them in RegulatedToolset:

from rgltr import boundary, governed, tag
from rgltr.pydantic_ai import RegulatedToolset

# Source: activates tags
@tag(activates=[Tag.CUSTOMERS])
async def get_customer(customer_id: str) -> dict:
    return await db.get_customer(customer_id)

@tag(activates=[Tag.FINANCIAL])
async def get_invoices(customer_id: str) -> list:
    return await db.get_invoices(customer_id)

# Boundary: visibility controlled by tags connected to boundary
@boundary(Boundary.EXTERNAL)
async def post_to_slack(message: str) -> None:
    await slack.post(message)

# Governed-only: audited/covered but no restrictions
@governed()
async def internal_healthcheck() -> str:
    return "ok"

# Tool with explicit blocked_by
@tag(blocked_by=[Tag.CUSTOMERS])
async def export_to_csv(data: dict) -> str:
    """Hidden when CUSTOMERS active."""
    return convert_to_csv(data)

toolset = RegulatedToolset([
    get_customer,
    get_invoices,
    post_to_slack,
    internal_healthcheck,
    export_to_csv,
])

agent = RegulatedAgent(Agent(..., toolsets=[toolset]), agent_id="app")

Using Tool / toolset decorators with rgltr

Pydantic AI supports both plain functions and explicit Tool objects (via FunctionToolset.tool(...) or Tool(...)). rgltr is compatible with either: our decorators only attach metadata (__rgltr__) and do not wrap the function.

Preferred usage: decorate your functions with rgltr (@tag, @boundary, @policy). When you use RegulatedAgent, rgltr automatically wraps all function tools registered via @agent.tool, @agent.tool_plain, Agent(..., tools=[...]), or FunctionToolset/toolsets you pass in.

When you still need RegulatedToolset: use it when you are not using RegulatedAgent (e.g., you want a governed toolset for tests, manual tool invocation, or non‑agent orchestration).

from pydantic_ai import FunctionToolset
from pydantic_ai.tools import Tool
from rgltr import tag
from rgltr.pydantic_ai import RegulatedToolset

@tag(activates=[Tag.CUSTOMERS])
async def get_customer(customer_id: str) -> dict:
    ...

toolset = RegulatedToolset([get_customer, Tool(get_customer, name="get_customer")])


toolset_decorator = FunctionToolset()

@toolset_decorator.tool(name="get_customer")
@tag(activates=[Tag.CUSTOMERS])
async def get_customer_v2(customer_id: str) -> dict:
    ...

toolset = RegulatedToolset(toolset=toolset_decorator)

Filtering Function Toolsets

Pydantic AI’s FilteredToolset can be used to prune the function tool list before the model sees it (feature flags, tenant tier, environment, etc.) while still applying rgltr’s tags/boundaries/policies to whatever remains.

from pydantic_ai.toolsets import FilteredToolset
from pydantic_ai.toolsets.function import FunctionToolset
from rgltr.pydantic_ai import RegulatedAgent

toolset = FunctionToolset()

@toolset.tool
async def internal_only(note: str) -> str:
    return f"note:{note}"

@toolset.tool
async def public_help(topic: str) -> str:
    return f"help:{topic}"

filtered = FilteredToolset(toolset, lambda _ctx, tool_def: tool_def.name != "internal_only")

agent = RegulatedAgent(Agent(..., toolsets=[filtered]), agent_id="support")

When to Use RegulatedToolset vs FilteredToolset

RegulatedToolset = governance (tags, boundaries, policies, audits).
FilteredToolset = pre‑LLM pruning (feature flags, tiers, environment, tool list size).

Typical patterns:

1. Governance only (using RegulatedAgent, no explicit RegulatedToolset):

agent = RegulatedAgent(Agent(..., tools=[...]), agent_id="app")

2. Filtering only (no rgltr):

agent = Agent(..., toolsets=[FilteredToolset(toolset, predicate)])

3. Filtering + governance (filter first, then rgltr governs):

toolset = FunctionToolset([...])
filtered = FilteredToolset(toolset, predicate)
agent = RegulatedAgent(Agent(..., toolsets=[filtered]), agent_id="app")

4. Governance without RegulatedAgent (manual/tooling use):

toolset = RegulatedToolset([get_customer, post_to_slack])
agent = Agent(..., toolsets=[toolset])

Policy Enforcement

For authorization beyond data flow—role checks. Policies run at call time, after visibility checks pass.

from datetime import datetime
from rgltr import policy, PolicyDecision, PolicyRequest

async def require_admin(request: PolicyRequest) -> PolicyDecision:
    user = request.run_context.deps
    if user and user.is_admin:
        return PolicyDecision.allow("user is admin")
    return PolicyDecision.deny("admin role required")

async def business_hours(request: PolicyRequest) -> PolicyDecision:
    if 9 <= datetime.now().hour < 17:
        return PolicyDecision.allow("within business hours")
    return PolicyDecision.deny("only available during business hours (9-17)")

async def large_batch_approval(request: PolicyRequest) -> PolicyDecision:
    count = request.args.get("count", 0)
    if count > 100:
        return PolicyDecision.requires_approval(
            f"Batch operation on {count} items requires approval"
        )
    return PolicyDecision.allow("small batch")

PolicyRequest structure:

@dataclass
class PolicyRequest:
    tool_name: str                  # Which tool
    args: dict     
    active_tags: set[Tag]           # Currently active tags
    boundary: Boundary | None       # Tool's boundary (None if no @boundary)
    agent_id: str                   # From RegulatedAgent
    run_id: str                     # Current run
    run_context: RunContext         # Pydantic AI run context (use .deps for app data)

Applying Policies to Function Tools

# All policies must pass (AND)
@boundary(Boundary.EXTERNAL)
@policy(require=[require_admin, business_hours])
async def delete_account(account_id: str) -> None:
    ...

# Any policy can allow (OR)
@policy(any_of=[require_admin, has_role("support")])
async def view_audit_log() -> list:
    ...

# Combined: business_hours AND (admin OR support)
@boundary(Boundary.EXTERNAL)
@policy(require=[business_hours], any_of=[require_admin, has_role("support")])
async def modify_settings(settings: dict) -> None:
    ...

Custom Error Messages

Optionally you can provide a consistent, user‑safe error message to help the LLM recover (and avoid leaking policy details). This overrides and unifies messages from multiple policies:

@policy(
    require=[require_admin],
    denied_message="Only administrators can delete accounts. Contact your admin."
)
async def delete_user(user_id: str) -> None:
    ...

Applying Policies to MCP Tools

regulated_mcp = RegulatedMCPAdapter(
    server=mcp,
    tools={
        "*": False,
        "issues_*": True,

        "repos_delete": [
            boundary(Boundary.EXTERNAL),
            policy(require=[require_admin]),
        ],

        "org_settings": [
            policy(require=[require_admin, business_hours]),
        ],
    },
)

Handling Policy Decisions

from rgltr import PolicyDenied, ApprovalRequired

try:
    result = await agent.run("Delete all inactive accounts", deps=current_user)

except PolicyDenied as e:
    log.warning(f"Denied: {e.request.tool_name} - {e.reason}")
    # LLM already received error message and may have recovered

except ApprovalRequired as e:
    reason = e.metadata.get("reason") if e.metadata else None
    if await show_approval_dialog(e.request.tool_name, reason):
        result = await handle_approved_action(e.request)
    else:
        result = "Action cancelled by user"

Configuration

rgltr supports two configuration scopes:

Scope	API	Use case
Global	configure(...)	Shared defaults for all agents
Per-agent	RegulatedAgent(..., config=...)	Agent-specific overrides

Global Configuration

from rgltr import configure

configure(
    ...
)

Per-Agent Configuration

from rgltr import RegulatorConfig
from rgltr.pydantic_ai import RegulatedAgent

admin_agent = RegulatedAgent(
    Agent(...),
    agent_id="admin-agent",
    config=RegulatorConfig(
      ...
    ),
)

Configuration Options

• boundaries (global / per-agent, default {}): Map boundaries to the tags that hide them (e.g., Boundary.EXTERNAL: True hides external tools when any tag is active).
• mode (global / per-agent, default "enforce"): "enforce" hides tools and blocks policy violations; "monitor" keeps tools visible and logs violations without blocking.
• require_annotations (global / per-agent, default False): Fail if a regulated tool lacks any rgltr decorators. Useful for ensuring every tool has been reviewed.
• audit_sink (global / per-agent, default AuditLogger()): Where audit events are written (set to None to disable logging).
• report_coverage (global / per-agent, default True): Emit a coverage report event at run start with all known tools and rules.

RegulatedAgent options:

• agent_id=None (default): audit grouping label. If omitted, rgltr uses Agent.name or generates an ID.
• wrap_agent_toolset=True (default): automatically govern tools registered via @agent.tool, @agent.tool_plain, tools=[...], and agent-level toolsets.
• govern_builtin_tools=False (default): opt-in gating for builtin tools (web search, code execution, etc.).

Audit Logging

Every tool call generates an audit event that can be logged. rgltr provides a simple file based logger that logs decisions without exposing tool arguments.

When running with configure(mode="monitor"), policy denials are logged but tools execute. Use audit logs to identify violations before switching to enforce mode by filtering for event_type="policy_decision" and enforced=false.

from rgltr import configure, AuditLogger

configure(audit_sink=AuditLogger(file="audit.jsonl"))

Event types:

Event	Description
tool_allowed	Tool executed successfully
tool_hidden	Tool was hidden from LLM (visibility)
policy_decision	Policy returned deny() or requires_approval()

Example output (JSONL):

{"ts":"2025-12-30T10:25:14.123000+00:00","ts_ms":1735554314123,"schema_version":1,"run_id":"abc-123","mode":"enforce","level":"info","event_type":"tool_allowed","tool_kind":"function","tool_name":"get_customer","tool_call_id":"call-1","agent_id":"customer-support","active_tags":[],"active_tags_after":["customers"]}
{"ts":"2025-12-30T10:25:15.001000+00:00","ts_ms":1735554315001,"schema_version":1,"run_id":"abc-123","mode":"enforce","level":"info","event_type":"tool_hidden","tool_kind":"function","tool_name":"post_to_slack","agent_id":"customer-support","active_tags":["customers"],"boundary":"Boundary.EXTERNAL","reason":"boundary"}
{"ts":"2025-12-30T10:25:17.220000+00:00","ts_ms":1735554317220,"schema_version":1,"run_id":"abc-123","mode":"monitor","level":"warn","event_type":"policy_decision","tool_kind":"function","tool_name":"delete_account","tool_call_id":"call-3","agent_id":"customer-support","active_tags":[],"decision":"deny","policy_reason":"admin required","enforced":false}
{"ts":"2025-12-30T10:25:17.221000+00:00","ts_ms":1735554317221,"schema_version":1,"run_id":"abc-123","mode":"monitor","level":"info","event_type":"tool_allowed","tool_kind":"function","tool_name":"delete_account","tool_call_id":"call-3","agent_id":"customer-support","active_tags":[],"active_tags_after":[],"would_block":true,"would_require_approval":false}

Custom audit sinks:

import json
from rgltr import AuditEvent, configure

def my_audit_handler(event: AuditEvent):
    print(json.dumps({
        "agent_id": event.agent_id,
        "run_id": event.run_id,
        "event_type": event.event_type,
        "tool_kind": event.tool_kind,
        "tool_name": event.tool_name,
        "tool_call_id": event.tool_call_id,
        "active_tags": event.active_tags,
        "active_tags_after": event.active_tags_after,
        "decision": event.decision,
        "enforced": event.enforced,
        "would_block": event.would_block,
        "would_require_approval": event.would_require_approval,
    }))

configure(audit_sink=my_audit_handler)

Coverage Reporting

configure(report_coverage=True)

Emits a coverage report at agent startup showing which tools have governance rules:

{
  "event": "coverage_report",
  "agent_id": "customer-support",
  "function_tools": {
    "get_customer": {
      "boundary": null,
      "policies": [],
      "tag": {"activates": ["customers"], "blocked_by": []},
    },
    "post_to_slack": {
      "boundary": "Boundary.EXTERNAL",
      "policies": [],
      "tag": {"activates": [], "blocked_by": []},
    }
  },
  "mcp_servers": {
    "context7": {
      "tools": {"*": false, "query-docs": true},
      "server_rules": {
        "boundary": null,
        "policies": [],
        "tag": {"activates": [], "blocked_by": []},
      },
      "active_tools": {"regulated": ["query-docs"], "non_regulated": []},
      "hidden_tools": [],
      "filtered_tools": ["resolve-library-id"]
    }
  },
  "boundaries": {
    "Boundary.EXTERNAL": ["docs"]
  }
}

Utility Scripts

• scripts/visibility_experiment.py: runs a simple two-turn experiment (tool visibility changes mid-run) and writes audit logs. Example:

  OPENAI_API_KEY=... python scripts/visibility_experiment.py \
    --model [model_id] \
    --runs 5 \
    --audit-file scripts/audit-visibility.jsonl \
    --retries 2

  This script requires network access and a model provider API key.

• scripts/audit_summary.py: summarizes an audit log (JSONL) produced by the experiment. Example:

  python scripts/audit_summary.py scripts/audit-visibility.jsonl

Complete Example

from enum import Enum
from pydantic_ai import Agent
from pydantic_ai.mcp import MCPServerStreamableHTTP
from rgltr import (
    tag, boundary, policy,
    PolicyDecision, PolicyRequest,
    configure, RegulatorConfig, AuditLogger
)
from rgltr.pydantic_ai import RegulatedAgent, RegulatedMCPAdapter, RegulatedToolset
import rgltr

# User-defined tags and boundaries
class Tag(Enum):
    CUSTOMERS = "customers"
    INTERNAL = "internal"

class Boundary(Enum):
    EXTERNAL = "external"
    USER = "user"

# Policy function
async def require_admin(request: PolicyRequest) -> PolicyDecision:
    user = request.run_context.deps
    if user and user.is_admin:
        return PolicyDecision.allow("admin")
    return PolicyDecision.deny("admin role required")

# Function tools
@tag(activates=[Tag.CUSTOMERS])
async def get_customer(customer_id: str) -> dict:
    return await db.get_customer(customer_id)

@boundary(Boundary.EXTERNAL)
async def post_to_slack(channel: str, message: str) -> None:
    await slack.post(channel, message)

toolset = RegulatedToolset([
    get_customer, post_to_slack
])

# MCP server with filtering + governance
mcp = MCPServerStreamableHTTP(url="https://mcp.example.com/github")

regulated_mcp = RegulatedMCPAdapter(
    server=mcp,
    # Server-level rules are applied to all tools
    activates=[Tag.INTERNAL],
    blocked_by=[Tag.INTERNAL],
    boundary=Boundary.EXTERNAL,

    # Tool-level rules add on top of server-level rules
    # Filtering uses fnmatch-style patterns, with more specific matches taking priority
    tools={
        "*": False,
        "issues_*": True,
        "issues_create": [boundary(Boundary.EXTERNAL)],
        "repos_delete": [
            boundary(Boundary.EXTERNAL),
            policy(require=[require_admin]),
        ],
    },
)

# Configure global defaults
configure(
    boundaries={Boundary.EXTERNAL: True},
    audit_sink=AuditLogger(file="audit.jsonl"),
)

# Create regulated agent
agent = RegulatedAgent(
    Agent(
        "openai:gpt-4",
        toolsets=[toolset, regulated_mcp],
    ),
    agent_id="customer-support",
)

# Run with context isolation
async with rgltr.context():
    result = await agent.run(
        "Look up customer 123 and post summary to Slack",
        deps=current_user,
    )