AgentResult

Result object from non-streaming agent execution

Overview

The AgentResult class encapsulates the complete result of an agent’s execution in non-streaming mode. It provides access to the updated thread, new messages, final output, structured output data, and execution telemetry.

Class Definition

@dataclass
class AgentResult:
    thread: Thread                    # Updated thread with new messages
    new_messages: List[Message]       # New messages added during execution
    content: Optional[str]            # Final assistant response content
    structured_data: Optional[BaseModel]  # Validated Pydantic model (if response_type used)
    validation_retries: int = 0       # Structured output retry count
    retry_history: Optional[List[Dict[str, Any]]] = None
    execution: ExecutionDetails       # Events, duration, tokens, and tool summaries

execution is appended after the existing optional fields so existing positional AgentResult(...) construction keeps the same argument order. In application code, prefer reading fields by name from the object returned by agent.run(...).

Properties

thread · Thread

The updated thread containing all messages including those added during this execution

new_messages · List[Message]

List of new messages added during this execution (excludes pre-existing messages)

content · Optional[str]

The final assistant response content. None if no assistant message was generated

structured_data · Optional[BaseModel]

When response_type is provided to agent.run(), this field contains a validated instance of the Pydantic model. Internally, the agent uses the output-tool pattern: your schema becomes a special tool, and when the LLM calls it, the arguments are validated against your model.

from pydantic import BaseModel

class Invoice(BaseModel):
    vendor: str
    total: float

result = await agent.run(thread, response_type=Invoice)
invoice: Invoice = result.structured_data
print(f"Vendor: {invoice.vendor}, Total: ${invoice.total}")

Returns None if response_type was not provided. See the Structured Output Guide for complete usage.

validation_retries · int

Number of structured output validation retry attempts. This is only relevant when response_type and retry configuration are used.

retry_history · Optional[List[Dict[str, Any]]]

Validation retry details, including attempt numbers, validation errors, and response previews.

execution · ExecutionDetails

Execution telemetry collected during this run, including:

• events: ordered ExecutionEvent objects
• duration_ms: total execution time
• total_tokens: total token usage reported by the LLM
• tool_calls: structured tool call summaries with name, arguments, result or error, duration, and success

success · bool

Property that returns True when the execution completed without any EXECUTION_ERROR events.

Usage Examples

Basic Usage

from tyler import Agent, Thread, Message

agent = Agent(name="MyAgent", purpose="To help users")
thread = Thread()
thread.add_message(Message(role="user", content="Hello!"))

# Execute and get result
result = await agent.run(thread)

# Access the response
print(f"Response: {result.content}")

Accessing Metrics

print(f"Success: {result.success}")
print(f"Tokens used: {result.execution.total_tokens}")
print(f"Duration: {result.execution.duration_ms:.0f}ms")

for tool_call in result.execution.tool_calls:
    print(f"{tool_call.tool_name}: {tool_call.result or tool_call.error}")

Working with Messages

# Access all new messages
for message in result.new_messages:
    print(f"{message.role}: {message.content}")
    
    # Check for tool calls
    if message.tool_calls:
        print(f"  Tool calls: {len(message.tool_calls)}")
    
    # Check metrics
    if message.metrics:
        tokens = message.metrics.get("usage", {})
        print(f"  Tokens: {tokens.get('total_tokens', 0)}")

Error Handling

try:
    result = await agent.run(thread)
    print(f"Response: {result.content}")
except Exception as e:
    print(f"Error: {e}")
    # The thread may still have partial messages
    if thread.messages:
        last_message = thread.messages[-1]
        print(f"Last message: {last_message.content}")

Thread Management

# The thread is updated in-place
original_message_count = len(thread.messages)
result = await agent.run(thread)
new_message_count = len(result.thread.messages)

print(f"Added {new_message_count - original_message_count} messages")

# You can also access the thread directly
assert result.thread is thread  # Same object, modified in-place

Common Patterns

Conversation Loop

async def chat_loop(agent: Agent, thread: Thread):
    while True:
        user_input = input("You: ")
        if user_input.lower() == 'quit':
            break
            
        thread.add_message(Message(role="user", content=user_input))
        result = await agent.run(thread)
        
        print(f"Assistant: {result.content}")
        print(f"(Took {result.execution.duration_ms:.0f}ms)")

Result Analysis

def analyze_result(result: AgentResult):
    """Analyze agent execution results"""
    stats = {
        "success": result.success,
        "duration_ms": result.execution.duration_ms,
        "tokens": result.execution.total_tokens,
        "tool_calls": len(result.execution.tool_calls),
        "messages_added": len(result.new_messages),
        "has_content": result.content is not None,
        "used_tools": bool(result.execution.tool_calls),
    }
    
    return stats

Persisting Results

from narrator import ThreadStore

store = ThreadStore()

# Execute agent
result = await agent.run(thread)

# Save the updated thread
await store.save(result.thread)

await store.update_metadata(
    thread_id=result.thread.id,
    metadata={
        "last_execution_ms": result.execution.duration_ms,
        "last_tokens_used": result.execution.total_tokens,
        "last_response": result.content,
        "last_success": result.success,
    }
)

Structured Output Usage

When using structured output, access the validated data through structured_data:

from pydantic import BaseModel, Field
from typing import List, Literal

class SupportTicket(BaseModel):
    priority: Literal["low", "medium", "high"]
    category: str
    summary: str = Field(max_length=500)
    requires_escalation: bool

# Run with response_type
result = await agent.run(thread, response_type=SupportTicket)

# Access structured data
if result.structured_data:
    ticket: SupportTicket = result.structured_data
    print(f"Priority: {ticket.priority}")
    print(f"Category: {ticket.category}")
    print(f"Needs escalation: {ticket.requires_escalation}")

# Raw content is still available
print(f"Raw JSON: {result.content}")

Handling Errors

When structured output validation fails after all retries, StructuredOutputError is raised:

from tyler import StructuredOutputError

try:
    result = await agent.run(thread, response_type=SupportTicket)
except StructuredOutputError as e:
    print(f"Failed: {e.message}")
    print(f"Validation errors: {e.validation_errors}")
    print(f"Last response: {e.last_response}")

See the Structured Output Guide for complete documentation.

See Also

• Thread - Conversation management
• Agent - The main agent class
• RetryConfig - Retry configuration for structured output
• StructuredOutputError - Error when validation fails
• ExecutionEvent - Individual execution events used by streaming and AgentResult.execution.events