ConnectOnion - Best Agent Framework for Python

Quick Start

Add event handlers to your agent in 3 simple steps:

main.py

from connectonion import Agent, after_llm

def log_llm_calls(agent):
    """Track LLM performance"""
    trace = agent.current_session['trace'][-1]
    if trace['type'] == 'llm_call':
        duration = trace['duration_ms']
        print(f"⚡ LLM call: {duration:.0f}ms")

agent = Agent(
    "assistant",
    tools=[search],
    on_events=[after_llm(log_llm_calls)]
)

agent.input("Search for Python")

Python REPL

Interactive

⚡ LLM call: 1204ms

⚡ LLM call: 831ms

"I found results for Python..."

Tip: Event handlers receive the agent instance, giving you full access to current_session, messages, trace, and more.

All Event Types

Here's when each event fires and what you can do with it:

after_user_input

Fires once per turn, after user input is added

main.py

def add_timestamp(agent):
    from datetime import datetime
    timestamp = datetime.now().strftime("%H:%M:%S")
    agent.current_session['messages'].append({
        'role': 'system',
        'content': f'Current time: {timestamp}'
    })

agent = Agent("assistant", on_events=[
    after_user_input(add_timestamp)
])

Python REPL

Interactive

# LLM now sees timestamp in context

# Useful for: time-aware agents, logging, session metadata

after_llm

Fires after each LLM response (multiple times per turn)

main.py

from connectonion import llm_do

def add_reflection(agent):
    """Add AI-generated reflection after tools execute"""
    trace = agent.current_session['trace']

    # Find recent tool executions
    recent_tools = []
    llm_count = 0
    for entry in reversed(trace):
        if entry.get('type') == 'llm_call':
            llm_count += 1
            if llm_count >= 2:
                break
        elif entry.get('type') == 'tool_execution':
            recent_tools.append(entry)

    if recent_tools:
        result = recent_tools[0]['result'][:200]
        reflection = llm_do(
            f"Reflect on this result: {result}",
            model="gpt-4o-mini"
        )
        # Inject as assistant message (safe timing after tools)
        agent.current_session['messages'].append({
            'role': 'assistant',
            'content': f"💭 {reflection}"
        })

agent = Agent("assistant", tools=[search], on_events=[
    after_llm(add_reflection)
])

Python REPL

Interactive

💭 The search results provide comprehensive information about AI...

# Useful for: reflection, chain-of-thought, meta-cognition

after_tool

Fires after each successful tool execution

main.py

def monitor_performance(agent):
    """Log slow tool executions"""
    trace = agent.current_session['trace'][-1]
    if trace['type'] == 'tool_execution':
        timing = trace['timing']
        if timing > 1000:  # Over 1 second
            tool_name = trace['tool_name']
            print(f"⚠️ Slow: {tool_name} took {timing/1000:.1f}s")

agent = Agent("assistant", tools=[search, analyze], on_events=[
    after_tool(monitor_performance)
])

Python REPL

Interactive

⚠️ Slow: analyze took 2.3s

# Useful for: performance monitoring, caching, optimization

on_error

Fires when tool execution fails or tool not found

main.py

def handle_errors(agent):
    """Custom error handling"""
    trace = agent.current_session['trace'][-1]
    if trace.get('status') in ('error', 'not_found'):
        error = trace.get('error', 'Unknown error')
        print(f"❌ Error: {error}")

        # Log to monitoring service
        # Add recovery instructions to messages
        # Implement retry logic

agent = Agent("assistant", tools=[api_call], on_events=[
    on_error(handle_errors)
])

Python REPL

Interactive

❌ Error: API rate limit exceeded

# Useful for: error logging, retry logic, fallback behavior

Combining Multiple Events

Use multiple event handlers together for comprehensive monitoring and control:

main.py

from connectonion import Agent, after_user_input, after_llm, after_tool, on_error
from datetime import datetime

def log_session_start(agent):
    print(f"📝 Session started at {datetime.now()}")

def track_llm(agent):
    trace = agent.current_session['trace'][-1]
    if trace['type'] == 'llm_call':
        print(f"⚡ LLM: {trace['duration_ms']:.0f}ms")

def track_tools(agent):
    trace = agent.current_session['trace'][-1]
    if trace['type'] == 'tool_execution':
        print(f"🔧 Tool: {trace['tool_name']}")

def handle_errors(agent):
    trace = agent.current_session['trace'][-1]
    print(f"❌ Error: {trace.get('error')}")

agent = Agent(
    "full_monitoring",
    tools=[search, analyze],
    on_events=[
        after_user_input(log_session_start),
        after_llm(track_llm),
        after_tool(track_tools),
        on_error(handle_errors)
    ]
)

agent.input("Search and analyze Python")

Python REPL

Interactive

📝 Session started at 2025-01-04 15:30:42

⚡ LLM: 1204ms

🔧 Tool: search

⚡ LLM: 831ms

🔧 Tool: analyze

⚡ LLM: 1142ms

"Analysis complete..."

Key Concepts

Event Handler Signature

All event handlers receive the agent instance:

main.py

def my_event_handler(agent: Agent) -> None:
    # Access agent state
    messages = agent.current_session['messages']
    trace = agent.current_session['trace']
    user_prompt = agent.current_session['user_prompt']
    iteration = agent.current_session['iteration']

    # Modify agent state
    messages.append({'role': 'system', 'content': 'Context'})

    # Access agent attributes
    tool_names = agent.list_tools()
    model = agent.llm.model

Python REPL

Interactive

# Event handlers are regular Python functions

# Full access to agent internals

# Can read AND modify agent state

Message Injection Timing

Important: Use after_llm to inject messages after tool execution:

❌ Don't use after_tool: Injecting messages during tool execution breaks the OpenAI message sequence (assistant → tool results)

✅ Use after_llm: Fires after all tool results are added to messages, safe for injection

Error Handling

Event handlers follow fail-fast principle:

main.py

def failing_event(agent):
    raise RuntimeError("Event failed")

agent = Agent("test", on_events=[
    after_llm(failing_event)
])

agent.input("test")  # Raises RuntimeError

Python REPL

Interactive

RuntimeError: Event failed

# Exceptions propagate - agents stop on event errors

# Design events to be robust or handle exceptions internally

Real-World Use Cases

1. Performance Monitoring Dashboard

main.py

class PerformanceMonitor:
    def __init__(self):
        self.metrics = {
            'llm_calls': 0,
            'tool_calls': 0,
            'total_llm_time': 0,
            'total_tool_time': 0,
            'errors': 0
        }

    def track_llm(self, agent):
        trace = agent.current_session['trace'][-1]
        if trace['type'] == 'llm_call':
            self.metrics['llm_calls'] += 1
            self.metrics['total_llm_time'] += trace['duration_ms']

    def track_tool(self, agent):
        trace = agent.current_session['trace'][-1]
        if trace['type'] == 'tool_execution':
            self.metrics['tool_calls'] += 1
            self.metrics['total_tool_time'] += trace['timing']

    def track_error(self, agent):
        self.metrics['errors'] += 1

    def report(self):
        print(f"LLM calls: {self.metrics['llm_calls']}")
        print(f"Avg LLM time: {self.metrics['total_llm_time'] / max(1, self.metrics['llm_calls']):.0f}ms")
        print(f"Tool calls: {self.metrics['tool_calls']}")
        print(f"Errors: {self.metrics['errors']}")

monitor = PerformanceMonitor()
agent = Agent("monitored", tools=[search], on_events=[
    after_llm(monitor.track_llm),
    after_tool(monitor.track_tool),
    on_error(monitor.track_error)
])

agent.input("Complex task...")
monitor.report()

Python REPL

Interactive

LLM calls: 3

Avg LLM time: 1245ms

Tool calls: 2

Errors: 0

2. Automatic Context Injection

main.py

def inject_company_context(agent):
    """Add company-specific context to every query"""
    agent.current_session['messages'].append({
        'role': 'system',
        'content': '''You are a customer support agent for Acme Corp.
        - Be friendly and professional
        - Reference our 30-day return policy
        - Escalate billing issues to finance team'''
    })

agent = Agent(
    "support_agent",
    tools=[search_knowledge_base, create_ticket],
    on_events=[after_user_input(inject_company_context)]
)

Python REPL

Interactive

# Every user query now includes company context

# LLM follows company policies automatically

# No need to repeat instructions in every prompt

3. Smart Retry Logic

main.py

class RetryHandler:
    def __init__(self, max_retries=3):
        self.max_retries = max_retries
        self.retry_count = {}

    def handle_error(self, agent):
        trace = agent.current_session['trace'][-1]
        tool_name = trace.get('tool_name')

        # Track retries
        if tool_name not in self.retry_count:
            self.retry_count[tool_name] = 0

        self.retry_count[tool_name] += 1

        if self.retry_count[tool_name] < self.max_retries:
            # Add retry instruction to messages
            agent.current_session['messages'].append({
                'role': 'system',
                'content': f'Previous {tool_name} failed. Try with different parameters.'
            })
            print(f"🔄 Retry {self.retry_count[tool_name]}/{self.max_retries}")
        else:
            print(f"❌ Max retries reached for {tool_name}")

retry_handler = RetryHandler()
agent = Agent("resilient", tools=[flaky_api], on_events=[
    on_error(retry_handler.handle_error)
])

Python REPL

Interactive

🔄 Retry 1/3

🔄 Retry 2/3

✓ Success on retry 2

API Reference

Event Wrapper Functions

after_user_input(func: Callable[[Agent], None]) → EventHandler

Wraps a function to fire after user input is added to session.

before_llm(func: Callable[[Agent], None]) → EventHandler

Wraps a function to fire before each LLM call.

after_llm(func: Callable[[Agent], None]) → EventHandler

Wraps a function to fire after each LLM response.

before_tool(func: Callable[[Agent], None]) → EventHandler

Wraps a function to fire before each tool execution.

after_tool(func: Callable[[Agent], None]) → EventHandler

Wraps a function to fire after each successful tool execution.

on_error(func: Callable[[Agent], None]) → EventHandler

Wraps a function to fire when tool execution fails or tool is not found.

Agent Constructor

Agent(name, tools, on_events: Optional[List[EventHandler]] = None, ...)

on_events: List of event handlers wrapped with event type functions

Best Practices

✅ Keep handlers simple: Each event handler should do one thing well. Compose multiple handlers for complex behavior.

✅ Use after_llm for message injection: This is the safe time to inject context after tool execution completes.

✅ Handle exceptions internally: If your event handler can fail, catch exceptions to prevent stopping the agent.

❌ Don't inject during tool execution: Using after_tool to inject messages breaks the tool calling message sequence.

❌ Don't do heavy computation: Event handlers run synchronously and block agent execution. Keep them fast.

Hook into agent lifecycle

6 Event Types

Quick Start

All Event Types

after_user_input

after_llm

after_tool

on_error

Combining Multiple Events

Key Concepts

Event Handler Signature

Message Injection Timing

Error Handling

Real-World Use Cases

1. Performance Monitoring Dashboard

2. Automatic Context Injection

3. Smart Retry Logic

API Reference

Event Wrapper Functions

Agent Constructor

Best Practices

Next Steps

@xray Debugging →

Auto-Debug →

Agent Reference →

Examples →