Skip to main content

​
Business Automation Examples

Examples showing how to automate common business processes using AI agents for document processing, customer onboarding, and workflow automation.

​
Email Processing Automation

Automate email categorization and intelligent responses: 
from erdo import Agent, state
from erdo.actions import llm, memory, utils
from erdo.conditions import IsSuccess, TextEquals, TextContains

email_processor = Agent(
    name="email_processor",
    description="Automatically processes and responds to emails"
)

# Classify incoming email
classify_step = email_processor.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Classify this email: {state.email_content}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "category": {"type": "string", "enum": ["support", "sales", "general"]},
                    "urgency": {"type": "string", "enum": ["low", "medium", "high"]},
                    "requires_response": {"type": "boolean"}
                }
            }
        }
    )
)

# Generate response
response_step = email_processor.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Generate a professional response to this {classify_step.output.category} email: {state.email_content}",
        system_prompt="You are a professional customer service representative."
    ),
    depends_on=classify_step
)

# Handle urgent emails
classify_step.on(
    IsSuccess() & TextEquals("urgency", "high"),
    utils.send_status(
        status="urgent_email",
        message="High priority email requires attention",
        priority="high"
    )
)

# Store interactions
response_step.on(
    IsSuccess(),
    memory.store(memory={
        "content": f"Email from {state.email_sender}: {response_step.output.response}",
        "type": "email_interaction",
        "tags": ["email", classify_step.output.category]
    })
)

​
Document Processing

Process and extract data from documents: 
document_processor = Agent(
    name="document_processor",
    description="Processes documents and extracts structured data"
)

# Analyze document
analysis_step = document_processor.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Analyze this document: {state.document_content}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "document_type": {"type": "string"},
                    "key_fields": {"type": "array"},
                    "confidence": {"type": "number"}
                }
            }
        }
    )
)

# Extract data
extraction_step = document_processor.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Extract structured data from this {analysis_step.output.document_type}: {state.document_content}",
        system_prompt="Extract accurate, structured information from the document."
    ),
    depends_on=analysis_step
)

# Store extracted data
extraction_step.on(
    IsSuccess(),
    memory.store(memory={
        "content": extraction_step.output.response,
        "type": "document_data",
        "tags": ["document", analysis_step.output.document_type]
    })
)

​
Customer Onboarding

Automate customer onboarding workflows: 
onboarding_agent = Agent(
    name="customer_onboarding",
    description="Automates customer onboarding process"
)

# Validate customer information
validation_step = onboarding_agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Validate customer information: {state.customer_data}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "is_valid": {"type": "boolean"},
                    "missing_fields": {"type": "array"},
                    "validation_score": {"type": "number"}
                }
            }
        }
    )
)

# Generate welcome message
welcome_step = onboarding_agent.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Generate personalized welcome message for: {state.customer_data}",
        system_prompt="Create a warm, professional welcome message."
    ),
    depends_on=validation_step
)

# Handle validation failures
validation_step.on(
    IsSuccess() & TextEquals("is_valid", "false"),
    utils.send_status(
        status="validation_failed",
        message="Customer data validation failed",
        missing_fields=validation_step.output.missing_fields
    )
)

# Complete successful onboarding
welcome_step.on(
    IsSuccess(),
    memory.store(memory={
        "content": f"Customer onboarded: {welcome_step.output.response}",
        "type": "customer_onboarding",
        "tags": ["onboarding", "completed", state.customer_id]
    })
)

​
Invoice Processing

Automate invoice processing and approval: 
invoice_processor = Agent(
    name="invoice_processor",
    description="Processes invoices and handles approvals"
)

# Extract invoice data
extract_step = invoice_processor.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Extract invoice details: {state.invoice_content}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "invoice_number": {"type": "string"},
                    "amount": {"type": "number"},
                    "vendor": {"type": "string"},
                    "due_date": {"type": "string"}
                }
            }
        }
    )
)

# Validate invoice
validate_step = invoice_processor.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Validate invoice data: {extract_step.output.response}",
        system_prompt="Check for completeness and accuracy of invoice data."
    ),
    depends_on=extract_step
)

# Handle high-value invoices
extract_step.on(
    IsSuccess() & GreaterThan("amount", "5000"),
    utils.send_status(
        status="high_value_invoice",
        message="High-value invoice requires manager approval",
        invoice_details=extract_step.output.response
    )
)

# Store processed invoice
validate_step.on(
    IsSuccess(),
    memory.store(memory={
        "content": validate_step.output.response,
        "type": "processed_invoice",
        "tags": ["invoice", "processed", extract_step.output.vendor]
    })
)

​
Lead Qualification

Qualify and score sales leads: 
lead_qualifier = Agent(
    name="lead_qualifier",
    description="Qualifies and scores sales leads"
)

# Analyze lead information
analysis_step = lead_qualifier.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Analyze this sales lead: {state.lead_data}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "qualification_score": {"type": "number"},
                    "interest_level": {"type": "string"},
                    "budget_range": {"type": "string"},
                    "timeline": {"type": "string"}
                }
            }
        }
    )
)

# Generate follow-up strategy
strategy_step = lead_qualifier.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Create follow-up strategy for lead: {analysis_step.output.response}",
        system_prompt="Provide specific, actionable follow-up recommendations."
    ),
    depends_on=analysis_step
)

# Handle high-quality leads
analysis_step.on(
    IsSuccess() & GreaterThan("qualification_score", "80"),
    utils.send_status(
        status="hot_lead",
        message="High-quality lead identified",
        lead_score=analysis_step.output.qualification_score
    )
)

# Store lead analysis
strategy_step.on(
    IsSuccess(),
    memory.store(memory={
        "content": strategy_step.output.response,
        "type": "lead_analysis",
        "tags": ["lead", "qualified", state.lead_source]
    })
)

​
Task Assignment

Automatically assign tasks based on workload and skills: 
task_assigner = Agent(
    name="task_assigner",
    description="Assigns tasks to team members based on workload and skills"
)

# Analyze task requirements
task_analysis = task_assigner.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Analyze task requirements: {state.task_description}",
        response_format={
            "Type": "json_schema",
            "Schema": {
                "type": "object",
                "properties": {
                    "required_skills": {"type": "array"},
                    "estimated_hours": {"type": "number"},
                    "priority": {"type": "string"},
                    "complexity": {"type": "string"}
                }
            }
        }
    )
)

# Find best assignee
assignment_step = task_assigner.step(
    llm.message(
        model="claude-sonnet-4",
        query=f"Find best team member for task: {task_analysis.output.response}, Available team: {state.team_data}",
        system_prompt="Consider skills, workload, and availability when assigning tasks."
    ),
    depends_on=task_analysis
)

# Notify assignee
assignment_step.on(
    IsSuccess(),
    utils.send_status(
        status="task_assigned",
        message=f"Task assigned to {assignment_step.output.assignee}",
        task_details=state.task_description
    )
)

​
Best Practices

​
Error Handling

Always include error handling for automation workflows: 
process_step.on(
    IsError(),
    utils.send_status(
        status="automation_error",
        message=f"Process failed: {process_step.error.message}",
        priority="high"
    )
)

​
Human Oversight

Include human review for critical decisions: 
# Flag for human review when confidence is low
analysis_step.on(
    IsSuccess() & LessThan("confidence", "0.7"),
    utils.send_status(
        status="human_review_required",
        message="Low confidence result needs human review"
    )
)

​
Audit Trail

Maintain audit trails for business processes: 
process_step.on(
    IsSuccess(),
    memory.store(memory={
        "content": f"Process completed: {process_step.output.response}",
        "type": "audit_log",
        "tags": ["audit", "completed", state.process_id],
        "extra": {
            "timestamp": "{{system.current_time}}",
            "user": state.user_id
        }
    })
)

​
Next Steps

Data Analysis Examples

Explore AI-powered data analysis and insights generation

Core Concepts

Learn about agents, steps, actions, and result handlers

SDK Reference

Complete Python SDK documentation and API reference

Pre-built Agents

Leverage ready-made automation solutions