Skip to main content
The Flokoa pydantic-ai integration lets you take any pydantic_ai.Agent you have already written and serve it as an A2A-compatible server without modifying your agent code. The PydanticAIAgentExecutor handles tool injection, model configuration, and request routing so you can focus on agent logic rather than infrastructure.

​
Installation

pip install "flokoa[pydantic-ai]"

​
Quick example

Define your pydantic-ai agent in a regular Python module and point flokoa run at it. The executor discovers and wraps the agent object automatically.
my_agent.py
from pydantic_ai import Agent

agent = Agent(
    model="openai:gpt-4o",
    system_prompt="You are a helpful assistant.",
)
Then start the A2A server: 
flokoa run --module my_agent:agent --framework pydantic-ai --port 8080
The server starts on http://localhost:8080 and is ready to accept A2A requests immediately.

​
PydanticAIAgentExecutor

flokoa.integrations.pydantic_ai.agent_executor.PydanticAIAgentExecutor wraps a pydantic_ai.Agent and implements the A2A AgentExecutor interface. You rarely need to instantiate it directly — flokoa run does that for you — but understanding what it provides helps you design agents that work well with the operator. 
from flokoa.integrations.pydantic_ai.agent_executor import PydanticAIAgentExecutor

executor = PydanticAIAgentExecutor(agent=agent)
What the executor provides:
  • Automatic tool injection — Tools defined as AgentTool CRDs are mounted at /etc/flokoa/tools/ inside the container and loaded automatically on each request.
  • TTL-based caching — Tool definitions and model configuration are cached (default: 60 seconds) to avoid re-reading the filesystem on every request. Control the TTL with the FLOKOA_CACHE_TTL_SECONDS environment variable.
  • Change detection — The executor compares file modification times before deciding whether to rebuild the FunctionToolset. If nothing changed, the cached toolset is reused at zero cost.
  • Explicit cache invalidation — Call executor.invalidate_caches() to force a full reload on the next request.
class PydanticAIAgentExecutor(FlokoaAgentExecutor):
    def __init__(
        self,
        agent: "PydanticAIAgent",
        cache: ConfigCache | None = None,
        toolset_factory: ToolsetFactory | None = None,
    ): ...

    def invalidate_caches(self) -> None: ...

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None: ...

​
Automatic tool injection

When your agent runs inside Kubernetes, the Flokoa operator mounts tool configurations from AgentTool CRDs as files at /etc/flokoa/tools/. The executor reads those files at request time and builds a FunctionToolset from them — you do not list tools in your agent code at all.
Tool injection is only active when the operator mounts the tool files. During local development with flokoa run, no files are mounted, so the agent runs with only the tools you define directly on the Agent object.
The executor rebuilds the toolset when it detects that a mounted file has changed (via mtime), or when the cache TTL expires. This means the operator can update an AgentTool CRD and the running agent will pick up the new tool without a pod restart.

​
OpenAPI tools

The pydantic-ai integration supports the openapi tool type. When the operator mounts an OpenAPI tool definition, the executor creates an OpenAPIToolset from the spec and adds its tools to the FunctionToolset used for that request. 
# This happens automatically inside the executor — shown for illustration
from flokoa.tools.openapi import OpenAPIToolset

toolset = OpenAPIToolset.from_tool_definition(tool_definition)
tools = toolset.get_tools()
You do not need to write this code yourself. Declare the tool as an AgentTool CRD and the executor handles the rest.

​
Model configuration

When an Agent CR references a Model CR, the operator writes the model configuration to the container at runtime. The executor reads that configuration and constructs the appropriate pydantic-ai Model and Provider objects before each request, so the agent always uses the operator-specified model without any code changes. 
# The executor calls these internally when model_config is available
from flokoa.integrations.pydantic_ai.model_factory import create_model, create_provider

provider = create_provider(model_config)
model = create_model(model_config, provider)
If no Model CR is referenced and no model is configured on the Agent object itself, the executor raises a ProviderNotConfiguredError before attempting to run the agent.

​
Running in Kubernetes

Build a container image with your agent and flokoa[pydantic-ai] installed, then declare an Agent CR with framework: pydantic-ai. The operator handles everything else. 
apiVersion: agent.flokoa.ai/v1alpha1
kind: Agent
metadata:
  name: my-pydantic-agent
spec:
  framework: pydantic-ai
  model:
    name: gpt-4o-model
  runtime:
    type: standard
    spec:
      container:
        name: agent
        image: ghcr.io/myorg/my-agent:v1.0.0
        ports:
          - containerPort: 8080
            name: http
Always set framework: pydantic-ai in the Agent CRD even if your container only runs one framework. The operator uses this field to tag traces, apply framework-specific health checks, and route model configurations to the correct provider factory.