adapters.pydantic_ai_adapter
Pydantic AI agent adapter.
Wraps a Pydantic AI Agent to implement the BaseAgent interface, enabling use with protocol adapters.
PydanticAIAdapter Objects
class PydanticAIAdapter(BaseAgent)
Adapter for Pydantic AI agents.
Wraps a pydantic_ai.Agent to provide a consistent interface for protocol adapters.
Example:
from pydantic_ai import Agent
pydantic_agent = Agent("openai:gpt-4o", instructions="...")
agent = PydanticAIAgent( agent=pydantic_agent, name="my_agent", description="A helpful assistant", )
response = await agent.run("Hello!", context)
__init__
def __init__(agent: Agent,
name: str = "pydantic_ai_agent",
description: str = "A Pydantic AI powered agent",
version: str = "1.0.0",
agent_id: str | None = None,
selected_mcp_servers: list[Any] | None = None,
non_mcp_toolsets: list[Any] | None = None,
codemode_builder: Any = None)
Initialize the Pydantic AI agent adapter.
Arguments:
agent- The Pydantic AI Agent instance.name- Agent name.description- Agent description.version- Agent version.agent_id- Unique identifier for usage tracking (defaults to name).selected_mcp_servers- List of MCP server selections to use.non_mcp_toolsets- List of non-MCP toolsets (e.g., codemode, skills) to always include.codemode_builder- Optional callable to rebuild codemode toolset when MCP servers change.
name
@property
def name() -> str
Get the agent's name.
description
@property
def description() -> str
Get the agent's description.
version
@property
def version() -> str
Get the agent's version.
agent_id
@property
def agent_id() -> str
Get the agent's unique identifier for usage tracking.
selected_mcp_server_ids
@property
def selected_mcp_server_ids() -> list[str]
Get the list of selected MCP server IDs.
codemode_enabled
@property
def codemode_enabled() -> bool
Check if codemode is currently enabled.
set_codemode_enabled
def set_codemode_enabled(enabled: bool) -> bool
Enable or disable codemode at runtime.
When enabling, builds a new CodemodeToolset using the codemode_builder. When disabling, removes the existing CodemodeToolset.
Arguments:
enabled- Whether to enable (True) or disable (False) codemode.
Returns:
True if the operation succeeded, False otherwise.
update_mcp_servers
def update_mcp_servers(servers: list[Any]) -> None
Update the list of selected MCP servers.
Also rebuilds Codemode toolset if a builder was provided, to ensure tool bindings reflect the current MCP server selection.
Arguments:
servers- New list of MCP server selections to use.
pydantic_agent
@property
def pydantic_agent() -> Agent
Get the underlying Pydantic AI agent.
get_tools
def get_tools() -> list[ToolDefinition]
Get the list of tools available to this agent.
run
async def run(prompt: str, context: AgentContext) -> AgentResponse
Run the agent with a prompt.
Arguments:
prompt- User prompt/message.context- Execution context.
Returns:
Complete agent response.
stream
async def stream(prompt: str,
context: AgentContext) -> AsyncIterator[StreamEvent]
Run the agent with streaming output.
Uses agent.run() with an event_stream_handler so that the
full agent graph executes (multiple model→tool→model rounds)
while text deltas are still yielded token-by-token.
run_stream() is intentionally not used here because it stops
the graph at the first output matching the output type, which
prevents multi-round tool usage.
Arguments:
prompt- User prompt/message.context- Execution context.
Yields:
Stream events as they are produced.
run_with_codemode
async def run_with_codemode(prompt: str, context: AgentContext,
codemode_executor: Any) -> AgentResponse
Run the agent with CodeMode for programmatic tool composition.
This variant allows the agent to use agent-codemode for executing code that composes multiple tools efficiently.
Arguments:
prompt- User prompt/message.context- Execution context.codemode_executor- CodeModeExecutor instance.
Returns:
Complete agent response.