Skip to content

2026-03-03

PydanticAI + Intended: Governing AI Agent Tools Step by Step

Developer Relations · Developer Experience

PydanticAI + Intended: Governing AI Agent Tools Step by Step

PydanticAI is one of the fastest-growing frameworks for building production AI agents in Python. It combines Pydantic's type safety with a clean agent abstraction that makes tool use straightforward. But like every agent framework, PydanticAI does not include built-in authorization. If an agent has access to a tool, it can call that tool without restriction.

This tutorial walks you through adding Intended authority governance to a PydanticAI agent. By the end, every tool call your agent makes will be evaluated against your policies, risk-scored, and recorded in an immutable audit trail.

What You Will Build

You will build a PydanticAI agent that manages cloud infrastructure. The agent can list servers, restart services, and modify security group rules. Without governance, the agent can do all three actions without limits. With Intended, the agent can list servers freely, restart services with automatic approval during business hours, and modify security groups only with human approval.

Prerequisites

You need Python 3.11 or later, a PydanticAI installation, and a Intended account with an API key. Sign up at meritt.run for a free tier account.

bash
pip install pydantic-ai meritt-sdk

Step 1: Define Your Agent and Tools

Start with a standard PydanticAI agent:

python
from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from dataclasses import dataclass

@dataclass
class InfraDeps:
    cloud_client: CloudClient
    environment: str

class ServerInfo(BaseModel):
    server_id: str
    status: str
    region: str

agent = Agent(
    "openai:gpt-4o",
    deps_type=InfraDeps,
    system_prompt="You are an infrastructure management assistant.",
)

@agent.tool
async def list_servers(ctx: RunContext[InfraDeps]) -> list[ServerInfo]:
    """List all servers in the current environment."""
    return await ctx.deps.cloud_client.list_servers(ctx.deps.environment)

@agent.tool
async def restart_service(
    ctx: RunContext[InfraDeps], server_id: str, service_name: str
) -> str:
    """Restart a service on a specific server."""
    await ctx.deps.cloud_client.restart_service(server_id, service_name)
    return f"Service {service_name} restarted on {server_id}"

@agent.tool
async def modify_security_group(
    ctx: RunContext[InfraDeps],
    group_id: str,
    rule_type: str,
    cidr: str,
    port: int,
) -> str:
    """Modify a security group rule."""
    await ctx.deps.cloud_client.modify_sg(group_id, rule_type, cidr, port)
    return f"Security group {group_id} updated"

This agent works. It can call all three tools. There is no governance.

Step 2: Create the Intended Guard

The Intended guard wraps your tools and intercepts every call:

python
from meritt_sdk import IntendedToolGuard

guard = IntendedToolGuard(
    api_key="mrt_your_api_key_here",
    org_id="org_your_org_id",
    agent_id="infra-management-agent",
    domain_pack="infrastructure",
    fail_mode="closed",
)

The `domain_pack="infrastructure"` tells Intended to use the infrastructure domain pack for risk scoring. This pack understands that security group modifications are higher risk than listing servers, and that production environments carry more risk than staging.

Step 3: Create a Protected Agent

Now rebuild the agent with Intended-protected tools. The key is using PydanticAI's tool decorator with the guard's `protect` method:

python
protected_agent = Agent(
    "openai:gpt-4o",
    deps_type=InfraDeps,
    system_prompt="You are an infrastructure management assistant.",
)

@protected_agent.tool
async def list_servers(ctx: RunContext[InfraDeps]) -> list[ServerInfo]:
    """List all servers in the current environment."""
    # Read-only operation, low risk
    decision = await guard.evaluate(
        intent="infrastructure.compute.list",
        params={"environment": ctx.deps.environment},
    )
    if not decision.allowed:
        return f"Blocked: {decision.reason}"

    return await ctx.deps.cloud_client.list_servers(ctx.deps.environment)

@protected_agent.tool
async def restart_service(
    ctx: RunContext[InfraDeps], server_id: str, service_name: str
) -> str:
    """Restart a service on a specific server."""
    decision = await guard.evaluate(
        intent="infrastructure.compute.restart",
        params={
            "server_id": server_id,
            "service_name": service_name,
            "environment": ctx.deps.environment,
        },
    )
    if not decision.allowed:
        return f"Blocked: {decision.reason}"

    await ctx.deps.cloud_client.restart_service(server_id, service_name)
    return f"Service {service_name} restarted on {server_id}"

@protected_agent.tool
async def modify_security_group(
    ctx: RunContext[InfraDeps],
    group_id: str,
    rule_type: str,
    cidr: str,
    port: int,
) -> str:
    """Modify a security group rule."""
    decision = await guard.evaluate(
        intent="infrastructure.network.security-group.modify",
        params={
            "group_id": group_id,
            "rule_type": rule_type,
            "cidr": cidr,
            "port": port,
            "environment": ctx.deps.environment,
        },
    )
    if decision.escalated:
        return f"Escalated for human review: {decision.escalation_id}"
    if not decision.allowed:
        return f"Blocked: {decision.reason}"

    await ctx.deps.cloud_client.modify_sg(group_id, rule_type, cidr, port)
    return f"Security group {group_id} updated (token: {decision.token_id})"

Notice the three different handling patterns. For list operations, a simple allow/deny check. For restart operations, the same pattern but with higher-risk parameters. For security group modifications, the code checks for escalation as a distinct outcome.

Step 4: Use the Convenience Wrapper

The pattern above is explicit but verbose. For simpler integration, use the `wrap_tool` convenience method that handles the decision logic automatically:

python
from meritt_sdk import IntendedToolGuard, ToolConfig

guard = IntendedToolGuard(
    api_key="mrt_your_api_key_here",
    org_id="org_your_org_id",
    agent_id="infra-management-agent",
    domain_pack="infrastructure",
)

auto_agent = Agent(
    "openai:gpt-4o",
    deps_type=InfraDeps,
    system_prompt="You are an infrastructure management assistant.",
)

@auto_agent.tool
@guard.wrap_tool(intent="infrastructure.compute.list")
async def list_servers(ctx: RunContext[InfraDeps]) -> list[ServerInfo]:
    """List all servers in the current environment."""
    return await ctx.deps.cloud_client.list_servers(ctx.deps.environment)

@auto_agent.tool
@guard.wrap_tool(intent="infrastructure.compute.restart")
async def restart_service(
    ctx: RunContext[InfraDeps], server_id: str, service_name: str
) -> str:
    """Restart a service on a specific server."""
    await ctx.deps.cloud_client.restart_service(server_id, service_name)
    return f"Service {service_name} restarted on {server_id}"

@auto_agent.tool
@guard.wrap_tool(intent="infrastructure.network.security-group.modify")
async def modify_security_group(
    ctx: RunContext[InfraDeps],
    group_id: str,
    rule_type: str,
    cidr: str,
    port: int,
) -> str:
    """Modify a security group rule."""
    await ctx.deps.cloud_client.modify_sg(group_id, rule_type, cidr, port)
    return f"Security group {group_id} updated"

The `wrap_tool` decorator handles the entire decision flow: submit intent, check decision, block or escalate if needed, pass through if allowed. The original tool function only executes if the authority decision is "allow."

Step 5: Configure Policies

In the Intended console or via the API, configure policies for your agent:

python
from meritt_sdk import PolicyBuilder

policies = PolicyBuilder(org_id="org_your_org_id")

# Allow read operations without restriction
policies.add_rule(
    intent_pattern="infrastructure.compute.list",
    decision="allow",
    description="Allow listing infrastructure resources",
)

# Allow restarts during business hours, escalate outside
policies.add_rule(
    intent_pattern="infrastructure.compute.restart",
    decision="allow",
    conditions={"time_window": "09:00-17:00", "days": "mon-fri"},
    fallback="escalate",
    description="Allow service restarts during business hours only",
)

# Always escalate security group changes
policies.add_rule(
    intent_pattern="infrastructure.network.security-group.*",
    decision="escalate",
    description="Require human approval for security group changes",
)

await policies.apply()

Step 6: Run and Observe

Run the agent:

python
async def main():
    deps = InfraDeps(
        cloud_client=CloudClient(),
        environment="production",
    )
    result = await protected_agent.run(
        "List all servers, then restart the nginx service on srv-001, "
        "and open port 443 on sg-prod-web",
        deps=deps,
    )
    print(result.data)

The agent will successfully list servers (low risk, auto-approved), restart nginx (approved if during business hours, escalated otherwise), and attempt to modify the security group (always escalated for human review). Each decision is recorded with full risk scores, policy evaluation details, and a signed token.

What You Get

Every tool call is now governed. The infrastructure domain pack scores risk based on the specific action, the target environment, and the agent's behavioral history. The audit trail records every decision with cryptographic proof. Your compliance team can verify any decision independently using the public key.

The agent's behavior is unchanged from its perspective. It still reasons, plans, and calls tools. The governance layer is transparent. The latency overhead is under 50ms per decision. The security improvement is the difference between "the agent can do anything it has a permission for" and "the agent can do what policy explicitly authorizes, with proof."

Install the SDK, configure your policies, and wrap your tools. Your PydanticAI agents are now operating under authority.