Skip to main content

​
Sync Module

The Sync module enables you to synchronize agents from Python code directly to the Erdo platform, eliminating the need for CLI commands in your Python workflows.

​
Installation & Setup

from erdo.sync import Sync
from erdo.config import Config

# Verify configuration
config = Config()
if not config.is_authenticated():
    print("Please run 'erdo login' first")

​
Basic Usage

​
Sync Single Agent

from erdo import Agent
from erdo.actions import llm
from erdo.sync import Sync

# Create your agent
agent = Agent(
    name="data_analyzer",
    description="Analyzes data and provides insights"
)

# Add steps
analysis_step = agent.step(
    llm.message(
        model="claude-sonnet-4",
        context="Analyze this data: {{dataset}}"
    )
)

# Sync to platform
result = Sync(agent)
print(f"Synced agent: {result.agent_name} -> {result.agent_key}")
print(f"Status: {result.status}")

​
Sync from File

from erdo.sync import Sync

# Sync agents from a Python file
results = Sync.from_file("agents/my_agent.py")

for result in results:
    if result.status == "success":
        print(f"✓ Synced: {result.agent_name} -> {result.agent_key}")
    else:
        print(f"✗ Failed: {result.agent_name} - {result.error}")

​
Sync from Directory

from erdo.sync import Sync

# Sync all agents in a directory
results = Sync.from_directory(
    directory_path="agents/",
    recursive=True,
    exclude_patterns=["*.pyc", "__pycache__", "*.log"]
)

print(f"Synced {len([r for r in results if r.status == 'success'])} agents")

​
API Reference

​
Sync Class

class Sync:
    def __init__(self, agent: Agent, source_file_path: Optional[str] = None):
        """Initialize Sync for a single agent."""
    
    @classmethod
    def from_file(cls, file_path: str, exclude_patterns: Optional[List[str]] = None) -> List[SyncResult]:
        """Sync all agents from a Python file."""
    
    @classmethod  
    def from_directory(cls, directory_path: str = ".", recursive: bool = True, exclude_patterns: Optional[List[str]] = None) -> List[SyncResult]:
        """Sync all agents from a directory."""

​
SyncResult

class SyncResult:
    agent_name: str           # Name of the agent
    agent_key: str           # Platform key (e.g., "erdo.data-analyzer")
    status: str              # "success" or "error"
    message: Optional[str]   # Success/error message
    error: Optional[str]     # Detailed error if sync failed
    file_path: Optional[str] # Source file path

​
Advanced Usage

​
Custom Configuration

from erdo.config import Config
from erdo.sync import Sync

# Custom configuration
config = Config(
    api_url="https://staging.erdo.ai",
    organization="my-org"
)

# Use with sync
sync = Sync(agent, config=config)
result = sync.sync_agent()

​
Error Handling

from erdo.sync import Sync, SyncError

try:
    result = Sync(agent)
    if result.status == "success":
        print(f"Agent synced: {result.agent_key}")
    else:
        print(f"Sync failed: {result.error}")
except SyncError as e:
    print(f"Sync error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

​
Batch Operations

from erdo.sync import Sync

# Prepare multiple agents
agents = [
    Agent(name="agent_1", description="First agent"),
    Agent(name="agent_2", description="Second agent"),
    Agent(name="agent_3", description="Third agent")
]

# Sync all agents
results = []
for agent in agents:
    result = Sync(agent)
    results.append(result)

# Report results
successful = [r for r in results if r.status == "success"]
failed = [r for r in results if r.status == "error"]

print(f"Successfully synced: {len(successful)}")
print(f"Failed: {len(failed)}")

​
File Discovery

The sync module automatically discovers and extracts agents from Python files:

​
Supported Patterns

# Direct agent assignment
my_agent = Agent(name="my_agent")

# Agent with steps
data_agent = Agent(name="data_agent")
step = data_agent.step(llm.message(...))

# Multiple agents in one file
agent_1 = Agent(name="first")
agent_2 = Agent(name="second")

​
Exclusion Patterns

# Common exclusion patterns
exclude_patterns = [
    "*.pyc",           # Compiled Python
    "__pycache__",     # Python cache
    "*.log",           # Log files
    "test_*.py",       # Test files
    ".*",              # Hidden files
    "node_modules/",   # Node modules
    ".git/",           # Git directory
]

results = Sync.from_directory(
    "agents/",
    exclude_patterns=exclude_patterns
)

​
Integration Examples

​
Jupyter Notebook

# Cell 1: Create agent
from erdo import Agent
from erdo.actions import llm
from erdo.sync import Sync

agent = Agent(name="notebook_agent")
step = agent.step(llm.message(
    model="claude-sonnet-4", 
    context="Process: {{data}}"
))

# Cell 2: Sync to platform
result = Sync(agent)
print(f"Agent available at: {result.agent_key}")

# Cell 3: Test immediately
from erdo.invoke import Invoke
test_result = Invoke.by_key(
    result.agent_key,
    parameters={"data": "sample input"}
)
print(test_result.data)

​
CI/CD Pipeline

#!/usr/bin/env python3
"""Deploy agents to production."""

import sys
from pathlib import Path
from erdo.sync import Sync

def deploy_agents():
    """Deploy all agents from agents/ directory."""
    results = Sync.from_directory("agents/")
    
    successful = [r for r in results if r.status == "success"]
    failed = [r for r in results if r.status == "error"]
    
    if failed:
        print("❌ Some agents failed to deploy:")
        for result in failed:
            print(f"  - {result.agent_name}: {result.error}")
        return False
    
    print(f"✅ Successfully deployed {len(successful)} agents:")
    for result in successful:
        print(f"  - {result.agent_name} -> {result.agent_key}")
    
    return True

if __name__ == "__main__":
    success = deploy_agents()
    sys.exit(0 if success else 1)

​
Development Workflow

from erdo import Agent
from erdo.sync import Sync
from erdo.invoke import Invoke

def develop_and_test_agent():
    """Complete development workflow."""
    
    # 1. Create agent
    agent = Agent(name="dev_agent")
    # ... add steps ...
    
    # 2. Sync to platform
    sync_result = Sync(agent)
    if sync_result.status != "success":
        print(f"Sync failed: {sync_result.error}")
        return
    
    print(f"Agent synced: {sync_result.agent_key}")
    
    # 3. Test immediately
    test_result = Invoke.by_key(
        sync_result.agent_key,
        parameters={"test_input": "development test"}
    )
    
    print(f"Test result: {test_result.status}")
    print(f"Output: {test_result.data}")

# Run development cycle
develop_and_test_agent()

​
Best Practices

  1. Descriptive Names: Use clear, descriptive agent names that reflect their purpose
  2. Error Handling: Always check sync results and handle errors gracefully
  3. File Organization: Keep agents in dedicated directories (e.g., agents/)
  4. Version Control: Commit agent files to version control for team collaboration
  5. Testing: Test agents immediately after syncing to catch issues early
  6. Batch Operations: Use directory sync for multiple agents to improve efficiency

​
Troubleshooting

​
Common Issues

Authentication Errors 
# Check authentication
from erdo.config import Config
config = Config()
if not config.is_authenticated():
    print("Run: erdo login")
Invalid Agent Definition 
# Ensure agent has required fields
agent = Agent(
    name="valid_name",  # Required: string, no spaces
    description="Clear description"  # Required
)
File Not Found 
from pathlib import Path

file_path = "agents/my_agent.py"
if not Path(file_path).exists():
    print(f"File not found: {file_path}")
Network Issues 
from erdo.sync import SyncError

try:
    result = Sync(agent)
except SyncError as e:
    if "timeout" in str(e).lower():
        print("Network timeout - try again")
    elif "connection" in str(e).lower():
        print("Connection failed - check network")