> ## Documentation Index
> Fetch the complete documentation index at: https://docs.hexgate.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Pydantic AI

> wrap_pydantic_agent — gates every tool through PolicyEnforcer.

<Note>
  New to the adapters? Start with the [wrapping overview and comparison
  table](/adapters/overview#how-framework-wrapping-works).
</Note>

<Info>
  **Compatible versions:** see the [compatibility
  table](/adapters/overview#compatible-framework-versions) for the verified
  `pydantic-ai-slim` range.
</Info>

`wrap_pydantic_agent` returns a `HexgatePydanticAgent` proxy backed by a clone of
the original agent whose tools are gated by a freshly built `PolicyEnforcer`.
Tools registered via the `Agent(...)` constructor or via `@agent.tool` /
`@agent.tool_plain` are all picked up. The `hexgate_context` is supplied **per call**, so a
single wrapped agent can serve many users concurrently — role resolution happens
at call time from the contextvar.

```python theme={null}
import asyncio
from dotenv import load_dotenv
from pydantic_ai import Agent

from hexgate.runtime import HexgateContext
from hexgate.adapters.pydantic_ai import wrap_pydantic_agent

INSTRUCTION = (
    "You are a DevOps assistant for a Kubernetes platform. Help engineers read "
    "service logs, restart services, and scale deployments. Pull the service "
    "name, replica count, and environment (dev/staging/prod) from the request "
    "and act directly — the policy layer gates sensitive actions."
)


async def main():
    load_dotenv()

    agent = Agent("openai:gpt-4o-mini", name="devops_agent", instructions=INSTRUCTION)

    @agent.tool_plain
    def read_logs(service: str, env: str) -> str:
        """Return recent log lines for `service` in `env` (dev/staging/prod)."""
        return f"(stub) {service}@{env}: 200 OK, 200 OK, WARN upstream slow"

    @agent.tool_plain
    def restart_service(service: str, env: str) -> str:
        """Restart `service` in `env`."""
        return f"(stub) restarted {service}@{env}"

    @agent.tool_plain
    def scale_deployment(service: str, replicas: int, env: str) -> str:
        """Scale `service` to `replicas` pods in `env`."""
        return f"(stub) scaled {service}@{env} to {replicas} replicas"

    agent = wrap_pydantic_agent(
        agent=agent,
        api_key="sk-...",  # or rely on HEXGATE_API_KEY
    )

    result = await agent.run(
        "Scale the checkout service to 5 replicas in staging.",
        hexgate_context=HexgateContext(
            user_id="engineer_1",
            user_roles=["operator"],
            session_id="session_1",
        ),
    )
    print(result.output)


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

## Under the hood

* `wrap_pydantic_agent` builds a `PolicySet`, constructs one `PolicyEnforcer`,
  reads tools off the agent's internal `_function_toolset`, copies each tool with
  an enforcer-gated `function_schema.call`, and returns a shallow-copied agent
  whose toolset holds those gated copies — your original `agent` is untouched, so
  it can be reused or wrapped again independently.
* Each invocation method on `HexgatePydanticAgent` (`run` / `run_sync` /
  `run_stream` / `iter`) takes `hexgate_context=` and opens a `HexgateContext` scope around the
  delegated `Agent` call. The contextvar is per-task, so concurrent `run` calls
  for different users do not see each other's policies.
* A non-allow `Decision` raises `ModelRetry(decision.as_error_message())`;
  pydantic\_ai surfaces it back to the model as a tool-result message —
  `[policy_denied]` / `[approval_required]` markers in the same shape as the
  OpenAI / Google adapters — instead of aborting the run.
* Identity propagation uses `propagate_attributes(...)` so Langfuse spans carry
  the caller identity. Global tracing is enabled via `Agent.instrument_all()` on
  construction.

## Runnable example

`examples/devops_pydantic_ai.py` — `wrap_pydantic_agent` (Pydantic AI) end-to-end.