SDK Overview

The Erdo Python SDK provides a powerful framework for building intelligent AI agents. With our agent-centric API design, you can create sophisticated automation workflows that combine language models, web research, code execution, and custom integrations using clean, readable code. Pure Python Workflows: The SDK now includes sync, invoke, and test modules, enabling you to work entirely in Python without needing the CLI.

Installation

pip install erdo

Complete Python Workflow

Build, sync, and invoke agents entirely in Python:

from erdo import Agent
from erdo.actions import llm, memory
from erdo.conditions import IsSuccess, GreaterThan
from erdo.sync import Sync
from erdo.invoke import Invoke

# 1. Define your agent
agent = Agent(
    name="email_assistant",
    description="Automatically responds to emails with context-aware replies"
)

response_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        context="Generate a professional response to this email: {{email_content}}"
    )
)

# Add result handlers
response_step.on(
    IsSuccess() & GreaterThan("confidence", "0.8"),
    memory.store(memory={
        "content": response_step.output.response,
        "type": "email_response",
        "tags": ["automated", "high-quality"]
    })
)

# 2. Sync to platform
sync_result = Sync(agent)
print(f"Synced agent: {sync_result.agent_key}")

# 3. Invoke the agent
invoke_result = Invoke.by_key(
    sync_result.agent_key,
    parameters={"email_content": "Hi, I need help with my order..."}
)
print(f"Response: {invoke_result.data}")

Quick Start

Here’s a simple agent that automates email responses:

from erdo import Agent
from erdo.actions import llm, memory
from erdo.conditions import IsSuccess, GreaterThan

# Create your agent
agent = Agent(
    name="email_assistant",
    description="Automatically responds to emails with context-aware replies"
)

response_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query="Generate a professional response to this email: {{email_content}}"
    )
)

# Add result handlers
response_step.on(
    IsSuccess() & GreaterThan("confidence", "0.8"),
    memory.store(memory={
        "content": "{{response_step.output.response}}",
        "type": "email_response",
        "tags": ["automated", "high-quality"]
    })
)

Core Components

Agents

The main orchestrators that define your automation workflow

Steps

Individual actions within your agent’s process

Actions

Building blocks that define what each step does

Result Handlers

Conditional logic for processing step outcomes

SDK Modules

Sync

Sync agents from Python code to the platform

Invoke

Execute agents programmatically with parameters

Test

Run agent tests and validations

API Patterns

The Erdo SDK uses an agent-centric approach:

Creating Steps

from erdo import Agent
from erdo.actions import llm

agent = Agent(name="my_agent")

step = agent.step(
    llm.message(model="claude-sonnet-4", query="{{query}}")
)

Key Features

Clean syntax: Simple step creation with intuitive methods
Intuitive parameter order: Action first, then optional parameters
Direct handlers: Use step.on() for immediate action chaining
Type-safe output access: Use step.output.field syntax
Condition operators: Use & (and), | (or), ~ (not) for combining conditions

Action Types

Erdo provides multiple action types for different use cases:

Language Model Actions (llm)

from erdo.actions import llm

# Simple message generation
response_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query="Analyze this data: {{input_data}}"
    )
)

# Structured output with validation
analysis_step = agent.step(
    llm.message(
        model="gpt-4",
        query="Extract key insights from {{document}}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "insights": {"type": "array"},
                    "confidence": {"type": "number"}
                }
            }
        }
    )
)

Memory Operations

from erdo.actions import memory

# Store information
store_step = agent.step(
    memory.store(memory=
        memory={
            "content": "{{analysis_result}}",
            "type": "customer_insight",
            "tags": ["analysis", "{{customer_id}}"]
        }
    )
)

# Search stored memories
search_step = agent.step(
    memory.search(
        query="{{search_term}}",
        limit=5,
        organization_scope="specific"
    )
)

Web Research & Data Collection

from erdo.actions import websearch, webparser

# Search the web
search_step = agent.step(
    websearch.search(
        query="{{topic}} latest research",
        language="en",
        country="us"
    )
)

# Parse web content
parse_step = agent.step(
    webparser.parse(
        url="{{target_url}}",
        selectors=["h1", "p", ".content"],
        extract_links=True
    )
)

Code Execution

from erdo.actions import codeexec

# Execute Python code
code_step = agent.step(
    codeexec.execute(
        code_files=[{
            "filename": "analysis.py",
            "content": """

import pandas as pd
import numpy as np

# Process the data

data = pd.read_csv('{{data_source}}')
result = data.describe()
print(result.to_json())
"""
}],
parameters={"data_source": "{{input_file}}"}
)
)

Utility Actions

from erdo.actions import utils

# Send status updates
status_step = agent.step(
    utils.send_status(
        status="processing",
        message="Analysis in progress...",
        data={"progress": "50%"}
    )
)

# Echo data for debugging
debug_step = agent.step(
    utils.echo(
        message="Debug info: {{debug_data}}",
        data={"timestamp": "{{current_time}}"}
    )
)

Code Execution with .exec Decorator

For steps that run Python code using external files, use the @agent.exec decorator:

Basic .exec Usage

from erdo import Agent
from erdo.types import PythonFile
from erdo.actions import memory
from erdo.conditions import IsSuccess, GreaterThan

agent = Agent(name="data_analyzer")

@agent.exec(
    code_files=[PythonFile(filename="analysis_files/analyze.py")]
)
def analyze_data():
    """Execute data analysis using external code files."""
    from analysis_files.analyze import analyze_data

    # Execute the analysis
    results = analyze_data(context)
    return results

# Handle results from .exec steps
analyze_data.on(
    IsSuccess() & GreaterThan("confidence", "0.8"),
    memory.store(memory={
        "content": analyze_data.output.insights,
        "type": "analysis",
        "tags": ["analysis", "high-confidence"]
    })
)

Multi-File Code Execution

@agent.exec(
    code_files=[
        PythonFile(filename="analysis_files/analyze.py"),
        PythonFile(filename="analysis_files/utils.py")
    ]
)
def analyze_with_utils():
    """Execute analysis using multiple code files."""
    from analysis_files.analyze import analyze_data
    from analysis_files.utils import prepare_data

    prepared_data = prepare_data(context.parameters.get("dataset", {}))
    results = analyze_data(context, prepared_data)
    return results

PythonFile Reference

from erdo.types import PythonFile

# Reference external Python files in your .exec steps
code_files = [
    PythonFile(filename="processors/main.py"),
    PythonFile(filename="processors/utils.py")
]

For detailed examples and best practices, see the Code Execution Guide.

Result Handlers

Handle step outcomes with conditional logic:

Direct Handlers with step.on()

from erdo.conditions import IsSuccess, GreaterThan, IsError
from erdo.actions import memory, utils

# Single action handler
analysis_step.on(
    IsSuccess() & GreaterThan("confidence", "0.8"),
    utils.send_status(status="high_confidence", message="Analysis highly reliable")
)

# Multiple action handler
analysis_step.on(
    IsSuccess() & GreaterThan("confidence", "0.9"),
    utils.send_status(status="excellent", message="Excellent analysis results"),
    memory.store(memory={
        "content": "{{analysis_step.output.insights}}",
        "type": "excellent_analysis"
    }),
    utils.echo(message="Analysis stored successfully")
)

Result Handlers with step.on()

# Store high-confidence analysis results
analysis_step.on(
    IsSuccess() & GreaterThan("confidence", "0.8"),
    memory.store(memory={
        "content": "{{analysis_step.output.insights}}",
        "type": "analysis_result",
        "tags": ["analysis", "high-confidence"],
        "extra": {"confidence": "{{analysis_step.output.confidence}}"}
    })
)

# Handle analysis failures
analysis_step.on(
    IsError(),
    utils.send_status(
        status="failed",
        message="Analysis failed: {{analysis_step.error.message}}"
    )
)

Dependencies and Flow Control

Sequential Steps

# Step 1: Data collection
collect_step = agent.step(
    websearch.search(query="{{topic}}")
)

# Step 2: Analysis (depends on collection)
analyze_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query="Analyze this data: {{collect_step.output.results}}"
    ),
    depends_on=collect_step
)

# Step 3: Report (depends on analysis)
report_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query="Create report from: {{analyze_step.output.insights}}"
    ),
    depends_on=analyze_step
)

Parallel Steps

# These steps run in parallel
sales_step = agent.step(
    codeexec.execute(code_files=[{"filename": "sales.py", "content": "process_sales()"}])
)

marketing_step = agent.step(
    codeexec.execute(code_files=[{"filename": "marketing.py", "content": "process_marketing()"}])
)

# Combine results after parallel processing
combine_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        query="Combine results: {{sales_step.output.data}}, {{marketing_step.output.data}}"
    ),
    depends_on=[sales_step, marketing_step]
)

Conditions and Logic

Basic Conditions

from erdo.conditions import (
    IsSuccess, IsError, GreaterThan, LessThan,
    TextContains, TextEquals, Not
)

# Success/Error conditions
step.on(IsSuccess(), handle_success_action)
step.on(IsError(), handle_error_action)

# Numeric comparisons
step.on(
    IsSuccess() & GreaterThan("score", "0.8"),
    handle_high_score_action
)

# Text conditions
step.on(
    IsSuccess() & TextContains("category", "urgent"),
    handle_urgent_action
)

Complex Conditions

# Combine conditions with logical operators
step.on(
    IsSuccess() &
    (GreaterThan("confidence", "0.8") | TextContains("priority", "high")) &
    ~TextContains("status", "draft"),
    memory.store(memory={
        "content": "{{step.output.data}}",
        "type": "priority_result"
    })
)

Complete Example

Here’s a comprehensive example showing all the key patterns:

from erdo import Agent
from erdo.actions import llm, memory, codeexec, utils
from erdo.conditions import IsSuccess, GreaterThan, IsError, TextContains

# Create agent
data_processor = Agent(
    name="data_processor",
    description="Processes and analyzes data with AI insights"
)

# Step 1: Search for context
search_step = data_processor.step(
    memory.search(query="{{query}}", limit=10)
)

# Step 2: Generate analysis code
code_gen_step = data_processor.step(
    llm.message(
        model="claude-sonnet-4",
        query="Generate Python code to analyze: {{query}}",
        context="{{search_step.output.memories}}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "code": {"type": "string"},
                    "explanation": {"type": "string"}
                }
            }
        }
    ),
    depends_on=search_step
)

# Step 3: Execute the code
execute_step = data_processor.step(
    codeexec.execute(
        code_files=[{
            "filename": "analysis.py",
            "content": "{{code_gen_step.output.code}}"
        }]
    ),
    depends_on=code_gen_step
)

# Handle successful execution
execute_step.on(
    IsSuccess() & ~TextContains("stderr", "error"),
    memory.store(memory={
        "content": "{{execute_step.output.stdout}}",
        "type": "analysis_result",
        "tags": ["analysis", "{{query}}"]
    })
)

# Handle execution with warnings using direct handler
execute_step.on(
    IsSuccess() & TextContains("stderr", "warning"),
    utils.send_status(
        status="completed_with_warnings",
        message="Analysis completed but with warnings"
    )
)

# Handle errors
execute_step.on(
    IsError(),
    utils.send_status(
        status="failed",
        message="Execution failed: {{execute_step.error.message}}"
    )
)

SDK Modules Reference

Sync Module

Sync agents from Python code to the Erdo platform:

from erdo.sync import Sync

# Sync a single agent
result = Sync(agent)
print(f"Agent synced: {result.agent_key}")

# Sync from file
results = Sync.from_file("my_agent.py")
for result in results:
    print(f"Synced: {result.agent_name} -> {result.agent_key}")

# Sync all agents in directory
results = Sync.from_directory("agents/")
print(f"Synced {len(results)} agents")

Invoke Module

Execute agents programmatically:

from erdo.invoke import Invoke

# Invoke by agent object
result = Invoke(agent, parameters={"query": "analyze this data"})
print(f"Result: {result.data}")

# Invoke by agent key
result = Invoke.by_key(
    "erdo.data-analyzer",
    parameters={"dataset": "sales_data.csv"}
)
print(f"Status: {result.status}, Data: {result.data}")

# Streaming mode (for real-time events)
result = Invoke.by_key(
    "erdo.data-analyzer",
    parameters={"dataset": "large_data.csv"},
    stream=True
)
for event in result.events:
    print(f"Event: {event.type} - {event.data}")

Test Module

Write fast, parallel agent tests:

from erdo import invoke
from erdo.test import text_contains, json_path_equals

# Write agent tests with agent_test_* prefix
def agent_test_basic_query():
    """Test basic agent invocation."""
    response = invoke(
        "my-agent",
        messages=[{"role": "user", "content": "Test query"}],
        mode="replay",  # Free after first run
    )

    assert response.success
    assert text_contains(str(response.result), "expected text")

# Run tests with CLI
# erdo agent-test tests/test_my_agent.py

# Or programmatically
from erdo.test import run_tests
exit_code = run_tests("tests/test_my_agent.py", verbose=True)

Best Practices

Use descriptive agent names for easy identification
Structure dependencies clearly to avoid circular references
Handle both success and error cases for robust workflows
Use type-safe output access with step.output.field
Combine conditions logically using &, |, ~ operators
Store important results in memory for future reference
Provide clear status updates for better user experience

Next Steps

CLI vs SDK

Compare CLI and SDK approaches for your workflow

Concepts

Learn core concepts and patterns

Examples

See real-world implementations

Actions Reference

Complete action documentation

API Reference

For detailed API documentation, including all available methods, parameters, and response formats, see our complete API reference (coming soon).

Support

GitHub Issues - Bug reports and feature requests
Documentation - Complete guides and tutorials

Get Started

Core Concepts

Testing & Validation

Examples & Use Cases

Python SDK

TypeScript SDK

CLI Tools

Pre-built Agents

Advanced

Integrations

​SDK Overview

​Installation

​Complete Python Workflow

​Quick Start

​Core Components

Agents

Steps

Actions

Result Handlers

​SDK Modules

Sync

Invoke

Test

​API Patterns

​Creating Steps

​Key Features

​Action Types

​Code Execution with .exec Decorator

​Basic .exec Usage

​Multi-File Code Execution

​PythonFile Reference

​Result Handlers

​Direct Handlers with step.on()

​Result Handlers with step.on()

​Dependencies and Flow Control

​Sequential Steps

​Parallel Steps

​Conditions and Logic

​Basic Conditions

​Complex Conditions

​Complete Example

​SDK Modules Reference

​Sync Module

​Invoke Module

​Test Module

​Best Practices

​Next Steps

CLI vs SDK

Concepts

Examples

Actions Reference

​API Reference

​Support

SDK Overview

Installation

Complete Python Workflow

Quick Start

Core Components

SDK Modules

API Patterns

Creating Steps

Key Features

Action Types

Code Execution with .exec Decorator

Basic .exec Usage

Multi-File Code Execution

PythonFile Reference

Result Handlers

Direct Handlers with step.on()

Result Handlers with step.on()

Dependencies and Flow Control

Sequential Steps

Parallel Steps

Conditions and Logic

Basic Conditions

Complex Conditions

Complete Example

SDK Modules Reference

Sync Module

Invoke Module

Test Module

Best Practices

Next Steps

API Reference

Support