Skip to main content

adapters.base

Base agent interface for agent-runtimes.

Provides an abstract base class that all agent implementations must follow, enabling consistent protocol adapter integration.

AgentContext Objects​

@dataclass
class AgentContext()

Context for agent execution.

Attributes:

  • session_id - Unique session identifier.
  • user_id - User identifier.
  • conversation_history - Previous messages in the conversation.
  • metadata - Additional context metadata.

ToolCall Objects​

@dataclass
class ToolCall()

A tool call requested by the agent.

Attributes:

  • id - Unique identifier for this tool call.
  • name - Tool name.
  • arguments - Tool arguments.

ToolResult Objects​

@dataclass
class ToolResult()

Result of a tool execution.

Attributes:

  • tool_call_id - ID of the tool call this is a result for.
  • result - The tool result data.
  • error - Error message if the tool failed.

success​

@property
def success() -> bool

Whether the tool execution was successful.

StreamEvent Objects​

@dataclass
class StreamEvent()

An event from the agent's streaming response.

Attributes:

  • type - Event type (text, tool_call, tool_result, done, error).
  • data - Event data.

type​

"text", "tool_call", "tool_result", "done", "error"

AgentResponse Objects​

@dataclass
class AgentResponse()

Complete response from an agent.

Attributes:

  • content - Text content of the response.
  • tool_calls - Any tool calls made during execution.
  • tool_results - Results of tool executions.
  • usage - Token usage information.
  • metadata - Additional response metadata.

ToolDefinition Objects​

@dataclass
class ToolDefinition()

Definition of a tool available to the agent.

Attributes:

  • name - Tool name.
  • description - Human-readable description.
  • input_schema - JSON Schema for input parameters.
  • output_schema - JSON Schema for output (optional).

BaseAgent Objects​

class BaseAgent(ABC)

Abstract base class for all agent implementations.

This provides a consistent interface for agents that can be used with different protocol adapters (ACP, AG-UI, A2A, etc.).

Example:

class MyAgent(BaseAgent): async def run(self, prompt, context):

Implementation

pass

def get_tools(self): return [ToolDefinition(name="my_tool", ...)]

run​

@abstractmethod
async def run(prompt: str, context: AgentContext) -> AgentResponse

Run the agent with a prompt.

Arguments:

  • prompt - User prompt/message.
  • context - Execution context with session and history.

Returns:

Complete agent response.

stream​

@abstractmethod
async def stream(prompt: str,
                 context: AgentContext) -> AsyncIterator[StreamEvent]

Run the agent with streaming output.

Arguments:

  • prompt - User prompt/message.

  • context - Execution context with session and history.

    Yields​

    StreamEvent Stream events as they are produced.

get_tools​

@abstractmethod
def get_tools() -> list[ToolDefinition]

Get the list of tools available to this agent.

Returns:

List of tool definitions.

name​

@property
@abstractmethod
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.

handle_tool_result​

async def handle_tool_result(
        context: AgentContext,
        tool_result: ToolResult) -> Optional[AgentResponse]

Handle a tool result and optionally continue execution.

Some agents may need to continue processing after receiving tool results. Override this method to implement that behavior.

Arguments:

  • context - Execution context.
  • tool_result - Result from tool execution.

Returns:

Optional continuation response.

initialize​

async def initialize() -> None

Initialize the agent.

Called once when the agent is first created. Override to perform async setup.

cleanup​

async def cleanup() -> None

Clean up agent resources.

Called when the agent is being shut down. Override to perform async cleanup.