Automation Tools

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

from aimatrix.tools.automation import PlaywrightTool

# Initialize Playwright with AI capabilities
browser = PlaywrightTool(
    browsers=["chromium", "firefox", "webkit"],
    headless=False,  # Set to True for server environments
    ai_enhanced=True  # Enable AI-powered element detection
)

# AI agent can control browsers naturally
agent = AIAgent(
    name="web-automation-agent",
    tools=[browser],
    instructions="""
    You are a web automation expert. You can:
    1. Navigate to websites and fill forms
    2. Extract data from web pages
    3. Perform complex multi-step workflows
    4. Handle dynamic content and SPAs
    """
)

# Example: Automated lead generation
await agent.execute_task(
    "Go to LinkedIn, search for 'AI engineers in San Francisco', "
    "extract contact information from the first 20 profiles, "
    "and save to a spreadsheet"
)

# Programmatic browser control
page = await browser.new_page()

# Navigate with AI-powered waiting
await page.goto("https://example.com")

# Find elements using AI vision (when CSS selectors fail)
submit_button = await page.find_element_by_ai_description(
    "blue submit button at the bottom of the form"
)
await submit_button.click()

# Extract structured data
product_data = await page.extract_data_by_ai({
    "product_name": "The main product title",
    "price": "The current price, including currency",
    "availability": "Whether the product is in stock",
    "reviews": "Customer review count and average rating"
})

# Handle dynamic content
await page.wait_for_ai_condition(
    "The loading spinner has disappeared and the content is fully loaded"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

# Multi-tab workflow automation
async def automated_research(topic: str):
    # Open multiple tabs for parallel research
    tabs = []
    search_engines = [
        "https://google.com",
        "https://bing.com", 
        "https://duckduckgo.com"
    ]
    
    for engine in search_engines:
        page = await browser.new_page()
        await page.goto(engine)
        
        # Search using AI to find search box
        search_box = await page.find_element_by_ai_description("main search input")
        await search_box.fill(topic)
        await search_box.press("Enter")
        
        tabs.append(page)
    
    # Collect results from all tabs
    all_results = []
    for tab in tabs:
        await tab.wait_for_load_state("networkidle")
        results = await tab.extract_search_results()
        all_results.extend(results)
    
    return all_results

# Form automation with smart field detection
async def fill_application_form(form_data: dict):
    await page.goto("https://example.com/application")
    
    # AI automatically maps form data to fields
    await page.fill_form_with_ai(form_data)
    
    # Handle file uploads
    if "resume" in form_data:
        file_input = await page.find_element_by_ai_description("resume upload field")
        await file_input.set_input_files(form_data["resume"])
    
    # Submit with confirmation handling
    await page.submit_form_with_ai_confirmation()

# Screenshot-based automation for complex UIs
async def handle_canvas_application():
    # Take screenshot for AI analysis
    screenshot = await page.screenshot()
    
    # AI identifies interactive elements in canvas/complex UI
    elements = await browser.analyze_screenshot_for_elements(
        screenshot, 
        element_types=["button", "input", "dropdown", "clickable"]
    )
    
    # Interact with elements using coordinates
    for element in elements:
        if element.type == "button" and "submit" in element.text.lower():
            await page.mouse.click(element.x, element.y)
            break

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

from aimatrix.tools.automation import SeleniumTool

# Initialize with AI enhancement
selenium = SeleniumTool(
    driver="chrome",
    ai_enhanced=True,
    implicit_wait=10
)

# Create AI-powered driver
driver = await selenium.create_ai_driver()

# Navigate and interact
await driver.get("https://legacy-system.com")

# Handle old-style forms with AI assistance
username_field = await driver.find_element_by_ai("username or email input field")
await username_field.send_keys("user@example.com")

password_field = await driver.find_element_by_ai("password input field")
await password_field.send_keys("secure_password")

login_button = await driver.find_element_by_ai("login or sign in button")
await login_button.click()

# Wait for page transition with AI
await driver.wait_for_ai_condition(
    "The page has loaded and shows the user dashboard"
)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

from aimatrix.tools.automation import ComputerVisionTool

# Initialize computer vision automation
cv_automation = ComputerVisionTool(
    screen_resolution="auto",
    confidence_threshold=0.8,
    ai_model="gpt-4-vision"
)

# AI agent can control desktop applications
desktop_agent = AIAgent(
    name="desktop-automation-agent",
    tools=[cv_automation],
    instructions="""
    You can control any desktop application by seeing the screen and interacting with it.
    Use natural language to describe what you want to do, and I'll translate that into
    precise screen interactions.
    """
)

# Example: Excel automation without Excel API
await desktop_agent.execute_task(
    "Open Excel, create a new spreadsheet, add column headers for "
    "Name, Email, Phone, and Company, then format them as a table"
)

# Programmatic screen automation
# Take screenshot and analyze
screenshot = await cv_automation.take_screenshot()
elements = await cv_automation.find_ui_elements(
    screenshot,
    element_types=["button", "textbox", "dropdown", "menu"]
)

# Click on specific element
calculator_icon = await cv_automation.find_element_by_description(
    "Calculator app icon on the taskbar"
)
await cv_automation.click(calculator_icon.center)

# Type text into applications
await cv_automation.wait_for_window("Calculator")
await cv_automation.type_text("123+456=")

# Verify results
result = await cv_automation.read_text_from_region(
    region=(100, 50, 300, 100)  # x, y, width, height
)
assert "579" in result

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

from aimatrix.tools.automation import WindowsAutomationTool

# Initialize Windows automation
windows_automation = WindowsAutomationTool(
    use_ui_automation=True,  # Use Windows UI Automation API
    fallback_to_vision=True  # Fall back to computer vision if needed
)

# Find and control Windows applications
# Start application
notepad = await windows_automation.start_application("notepad.exe")

# Find window by title pattern
window = await windows_automation.find_window(title_pattern="*Notepad*")
await window.bring_to_front()

# Use UI Automation to find controls
text_area = await window.find_element(
    control_type="Edit",
    automation_id="15"  # Notepad's main text area
)

# Type text with proper formatting
await text_area.set_text("""
Meeting Notes - January 20, 2025
================================

Attendees:
- John Smith (CEO)
- Jane Doe (CTO)
- AI Assistant

Action Items:
1. Review Q1 budget proposals
2. Finalize product roadmap
3. Schedule team building event
""")

# Save file with dialog automation
await window.send_keys("^s")  # Ctrl+S

# Handle Save As dialog
save_dialog = await windows_automation.wait_for_window(
    title_pattern="Save As*",
    timeout=5
)

filename_field = await save_dialog.find_element(
    control_type="Edit",
    name="File name:"
)
await filename_field.set_text("meeting_notes_2025_01_20.txt")

save_button = await save_dialog.find_element(
    control_type="Button",
    name="Save"
)
await save_button.click()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

from aimatrix.tools.automation import MacOSAutomationTool

# Initialize macOS automation
macos_automation = MacOSAutomationTool(
    use_applescript=True,
    use_accessibility=True
)

# AppleScript integration for system-level tasks
applescript_code = '''
tell application "Mail"
    activate
    make new outgoing message with properties {
        subject: "Automated Report",
        content: "This is an automated report generated by AI.",
        visible: true
    }
    
    tell the front outgoing message
        make new to recipient at end of to recipients with properties {
            address: "manager@example.com"
        }
    end tell
end tell
'''

await macos_automation.run_applescript(applescript_code)

# Accessibility API for precise control
# Find and control UI elements
safari = await macos_automation.get_application("Safari")
if not safari.is_running():
    await safari.launch()

# Wait for Safari to be ready
await macos_automation.wait_for_application_ready("Safari")

# Get the main window
window = safari.windows[0]

# Find address bar using accessibility
address_bar = await window.find_element(
    role="AXTextField",
    description="Address and search bar"
)

# Navigate to website
await address_bar.set_value("https://example.com")
await address_bar.perform_action("AXConfirm")  # Press Enter

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

from aimatrix.tools.automation import APIOrchestrationTool

# Initialize API orchestration
api_tool = APIOrchestrationTool(
    rate_limiting=True,
    retry_logic=True,
    authentication_manager=True
)

# Define API workflow
@api_tool.workflow("customer_onboarding")
async def onboard_customer(customer_data: dict):
    """Complete customer onboarding across multiple systems"""
    
    # Step 1: Create customer in CRM
    crm_response = await api_tool.post(
        url="https://api.crm.com/customers",
        headers={"Authorization": f"Bearer {crm_token}"},
        json={
            "name": customer_data["name"],
            "email": customer_data["email"],
            "company": customer_data["company"]
        }
    )
    
    customer_id = crm_response.json()["id"]
    
    # Step 2: Set up billing account
    billing_response = await api_tool.post(
        url="https://api.billing.com/accounts",
        json={
            "customer_id": customer_id,
            "plan": customer_data["plan"],
            "payment_method": customer_data["payment_method"]
        }
    )
    
    # Step 3: Create user account
    user_response = await api_tool.post(
        url="https://api.platform.com/users",
        json={
            "customer_id": customer_id,
            "email": customer_data["email"],
            "role": "admin"
        }
    )
    
    # Step 4: Send welcome email
    await api_tool.post(
        url="https://api.email.com/send",
        json={
            "to": customer_data["email"],
            "template": "welcome_template",
            "variables": {
                "customer_name": customer_data["name"],
                "login_url": f"https://platform.com/login?user={user_response.json()['id']}"
            }
        }
    )
    
    return {
        "customer_id": customer_id,
        "billing_account": billing_response.json()["account_id"],
        "user_id": user_response.json()["id"]
    }

# Execute workflow with error handling
try:
    result = await onboard_customer({
        "name": "John Doe",
        "email": "john@example.com",
        "company": "Example Corp",
        "plan": "enterprise",
        "payment_method": "pm_123456789"
    })
    print(f"Customer onboarded successfully: {result}")
except APIWorkflowError as e:
    print(f"Onboarding failed at step {e.step}: {e.message}")
    # Implement rollback logic
    await rollback_onboarding(e.completed_steps)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

from aimatrix.tools.automation import GraphQLTool

# Initialize GraphQL client
graphql = GraphQLTool(
    endpoint="https://api.github.com/graphql",
    headers={"Authorization": f"Bearer {github_token}"}
)

# AI-powered query building
@graphql.ai_query("Get repository information")
async def get_repo_info(owner: str, name: str):
    """Get comprehensive repository information including issues, PRs, and commits"""
    
    # AI automatically builds optimized GraphQL query
    query = await graphql.build_query_with_ai(
        description="Get repository basic info, latest 10 issues, latest 10 PRs, and commit count",
        variables={"owner": owner, "name": name}
    )
    
    # Generated query might look like:
    # query($owner: String!, $name: String!) {
    #   repository(owner: $owner, name: $name) {
    #     name
    #     description
    #     stargazerCount
    #     forkCount
    #     issues(first: 10, states: OPEN) {
    #       nodes {
    #         title
    #         createdAt
    #         author { login }
    #       }
    #     }
    #     pullRequests(first: 10, states: OPEN) {
    #       nodes {
    #         title
    #         createdAt
    #         author { login }
    #       }
    #     }
    #     defaultBranchRef {
    #       target {
    #         ... on Commit {
    #           history(first: 1) {
    #             totalCount
    #           }
    #         }
    #       }
    #     }
    #   }
    # }
    
    result = await graphql.execute(query, variables={
        "owner": owner,
        "name": name
    })
    
    return result

# Use in automation workflow
repo_data = await get_repo_info("microsoft", "vscode")
print(f"VS Code has {repo_data['repository']['stargazerCount']} stars")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

from aimatrix.tools.automation import WebhookTool

# Initialize webhook manager
webhooks = WebhookTool(
    base_url="https://your-domain.com/webhooks",
    security_enabled=True
)

# Register webhook handlers
@webhooks.handler("stripe.payment.succeeded")
async def handle_successful_payment(event):
    """Handle successful payment from Stripe"""
    payment_data = event["data"]["object"]
    
    # Update customer billing status
    await api_tool.post(
        url="https://api.crm.com/customers/update-billing",
        json={
            "customer_id": payment_data["metadata"]["customer_id"],
            "status": "paid",
            "amount": payment_data["amount"],
            "payment_date": payment_data["created"]
        }
    )
    
    # Send confirmation email
    await api_tool.post(
        url="https://api.email.com/send",
        json={
            "to": payment_data["billing_details"]["email"],
            "template": "payment_confirmation",
            "variables": {
                "amount": payment_data["amount"] / 100,  # Convert from cents
                "currency": payment_data["currency"].upper()
            }
        }
    )

@webhooks.handler("github.push")
async def handle_code_push(event):
    """Handle code push to trigger CI/CD"""
    if event["ref"] == "refs/heads/main":
        # Trigger deployment
        await api_tool.post(
            url="https://api.ci-cd.com/trigger",
            json={
                "repository": event["repository"]["full_name"],
                "commit": event["head_commit"]["id"],
                "branch": "main"
            }
        )

# Start webhook server
await webhooks.start_server(port=8080)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

from aimatrix.tools.automation import WorkflowBuilder

# Initialize workflow builder
workflow_builder = WorkflowBuilder()

# Create workflow programmatically
customer_support_workflow = workflow_builder.create_workflow(
    name="customer_support_escalation",
    description="Automated customer support ticket escalation"
)

# Define workflow steps
start_node = workflow_builder.add_trigger(
    workflow=customer_support_workflow,
    trigger_type="email_received",
    conditions={"to": "support@company.com"}
)

sentiment_analysis = workflow_builder.add_action(
    workflow=customer_support_workflow,
    action_type="analyze_sentiment",
    inputs={"text": "{{trigger.email_body}}"},
    previous_node=start_node
)

decision_node = workflow_builder.add_decision(
    workflow=customer_support_workflow,
    condition="{{sentiment_analysis.score}} < -0.5",
    previous_node=sentiment_analysis
)

# High priority path (negative sentiment)
high_priority_branch = workflow_builder.add_action(
    workflow=customer_support_workflow,
    action_type="create_ticket",
    inputs={
        "title": "{{trigger.email_subject}}",
        "description": "{{trigger.email_body}}",
        "priority": "high",
        "assignee": "senior_support_agent"
    },
    condition_branch="true",
    previous_node=decision_node
)

notify_manager = workflow_builder.add_action(
    workflow=customer_support_workflow,
    action_type="send_slack_message",
    inputs={
        "channel": "#support-escalation",
        "message": "High priority ticket created: {{high_priority_branch.ticket_url}}"
    },
    previous_node=high_priority_branch
)

# Normal priority path
normal_priority_branch = workflow_builder.add_action(
    workflow=customer_support_workflow,
    action_type="create_ticket",
    inputs={
        "title": "{{trigger.email_subject}}",
        "description": "{{trigger.email_body}}",
        "priority": "normal",
        "assignee": "support_agent"
    },
    condition_branch="false",
    previous_node=decision_node
)

auto_reply = workflow_builder.add_action(
    workflow=customer_support_workflow,
    action_type="send_email",
    inputs={
        "to": "{{trigger.email_from}}",
        "subject": "Re: {{trigger.email_subject}}",
        "body": "Thank you for contacting us. We've received your request (Ticket #{{normal_priority_branch.ticket_id}}) and will respond within 24 hours."
    },
    previous_node=normal_priority_branch
)

# Deploy workflow
await workflow_builder.deploy(customer_support_workflow)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

# Complex conditional workflow
@workflow_builder.workflow("invoice_processing")
async def process_invoice(invoice_data: dict):
    """Automated invoice processing with multiple validation steps"""
    
    # Step 1: Validate invoice data
    validation_result = await workflow_builder.validate_data(
        data=invoice_data,
        schema={
            "amount": {"type": "number", "minimum": 0},
            "vendor": {"type": "string", "required": True},
            "due_date": {"type": "date", "future": True}
        }
    )
    
    if not validation_result.is_valid:
        # Send back for correction
        await workflow_builder.send_notification(
            type="email",
            to=invoice_data["submitter_email"],
            template="invoice_correction_needed",
            data={"errors": validation_result.errors}
        )
        return {"status": "rejected", "reason": "validation_failed"}
    
    # Step 2: Check if vendor exists
    vendor_exists = await workflow_builder.check_condition(
        condition=f"vendor_exists('{invoice_data['vendor']}')"
    )
    
    if not vendor_exists:
        # Create new vendor workflow
        vendor_creation = await workflow_builder.trigger_sub_workflow(
            workflow="create_vendor",
            inputs={"name": invoice_data["vendor"]}
        )
        await vendor_creation.wait_for_completion()
    
    # Step 3: Amount-based approval routing
    if invoice_data["amount"] > 10000:
        # High-value invoice - requires CFO approval
        approval = await workflow_builder.request_approval(
            approver="cfo@company.com",
            subject=f"Approval needed: ${invoice_data['amount']} invoice from {invoice_data['vendor']}",
            timeout_hours=48
        )
        
        if approval.status != "approved":
            return {"status": "rejected", "reason": "not_approved"}
    
    elif invoice_data["amount"] > 1000:
        # Mid-value invoice - requires manager approval
        approval = await workflow_builder.request_approval(
            approver="manager@company.com",
            subject=f"Approval needed: ${invoice_data['amount']} invoice from {invoice_data['vendor']}",
            timeout_hours=24
        )
        
        if approval.status != "approved":
            return {"status": "rejected", "reason": "not_approved"}
    
    # Step 4: Process payment
    try:
        payment_result = await workflow_builder.execute_action(
            action="process_payment",
            inputs={
                "amount": invoice_data["amount"],
                "vendor": invoice_data["vendor"],
                "due_date": invoice_data["due_date"]
            },
            retry_attempts=3
        )
        
        # Step 5: Update accounting system
        await workflow_builder.execute_action(
            action="update_accounting",
            inputs={
                "invoice_id": invoice_data["id"],
                "payment_id": payment_result.payment_id,
                "status": "paid"
            }
        )
        
        return {"status": "completed", "payment_id": payment_result.payment_id}
        
    except PaymentError as e:
        # Handle payment failure
        await workflow_builder.send_notification(
            type="slack",
            channel="#finance-alerts",
            message=f"Payment failed for invoice {invoice_data['id']}: {str(e)}"
        )
        
        # Schedule retry
        await workflow_builder.schedule_retry(
            workflow="process_invoice",
            inputs=invoice_data,
            delay_hours=24
        )
        
        return {"status": "payment_failed", "retry_scheduled": True}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

from aimatrix.tools.automation import SchedulerTool

# Initialize scheduler
scheduler = SchedulerTool(
    timezone="UTC",
    persistent=True  # Survives system restarts
)

# Daily report generation
@scheduler.daily(hour=9, minute=0)  # 9:00 AM daily
async def generate_daily_report():
    """Generate and send daily business reports"""
    
    # Collect data from multiple sources
    sales_data = await api_tool.get("https://api.crm.com/sales/daily")
    support_data = await api_tool.get("https://api.support.com/metrics/daily")
    finance_data = await api_tool.get("https://api.finance.com/summary/daily")
    
    # Generate report with AI
    report = await ai_agent.generate_report(
        title="Daily Business Summary",
        data={
            "sales": sales_data.json(),
            "support": support_data.json(),
            "finance": finance_data.json()
        },
        format="html"
    )
    
    # Send to stakeholders
    await api_tool.post(
        url="https://api.email.com/send",
        json={
            "to": ["ceo@company.com", "cfo@company.com", "cmo@company.com"],
            "subject": f"Daily Report - {datetime.now().strftime('%Y-%m-%d')}",
            "html": report.content,
            "attachments": [report.pdf_version]
        }
    )

# Weekly cleanup tasks
@scheduler.weekly(day="sunday", hour=2, minute=0)  # 2:00 AM on Sundays
async def weekly_maintenance():
    """Perform weekly system maintenance tasks"""
    
    # Clean up temporary files
    await api_tool.delete("https://api.storage.com/temp-files/cleanup")
    
    # Archive old logs
    await api_tool.post("https://api.logs.com/archive", json={
        "older_than_days": 30
    })
    
    # Update customer engagement scores
    await workflow_builder.trigger_workflow(
        "update_customer_engagement_scores"
    )
    
    # Generate weekly analytics
    await workflow_builder.trigger_workflow(
        "generate_weekly_analytics"
    )

# Event-driven workflow
@scheduler.on_event("customer.signup")
async def handle_customer_signup(event_data):
    """Handle new customer signup"""
    
    # Wait 5 minutes to let initial setup complete
    await scheduler.delay(minutes=5)
    
    # Send welcome email sequence
    await workflow_builder.trigger_workflow(
        "customer_welcome_sequence",
        inputs={"customer_id": event_data["customer_id"]}
    )
    
    # Schedule follow-up check
    await scheduler.schedule_once(
        function=check_customer_activation,
        delay_days=3,
        args=[event_data["customer_id"]]
    )

# Custom cron schedule
@scheduler.cron("0 */6 * * *")  # Every 6 hours
async def sync_external_data():
    """Sync data from external partners"""
    
    partners = ["partner_a", "partner_b", "partner_c"]
    
    for partner in partners:
        try:
            # Sync data
            sync_result = await api_tool.post(
                url=f"https://api.{partner}.com/sync",
                headers={"Authorization": f"Bearer {get_partner_token(partner)}"}
            )
            
            # Process synced data
            if sync_result.status_code == 200:
                await workflow_builder.trigger_workflow(
                    "process_partner_data",
                    inputs={
                        "partner": partner,
                        "data": sync_result.json()
                    }
                )
                
        except Exception as e:
            # Log error and continue with other partners
            logger.error(f"Sync failed for {partner}: {str(e)}")
            
            # Send alert for critical partners
            if partner in ["critical_partner_a", "critical_partner_b"]:
                await workflow_builder.send_alert(
                    type="slack",
                    channel="#data-alerts",
                    message=f"Critical partner sync failed: {partner} - {str(e)}"
                )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

from aimatrix.tools.automation import NLToAPITool

# Initialize natural language API translator
nl_api = NLToAPITool(
    api_catalog="https://api-catalog.company.com",  # Your API documentation
    model="gpt-4"
)

# AI agent can execute natural language instructions
automation_agent = AIAgent(
    name="nl-automation-agent",
    tools=[nl_api, api_tool],
    instructions="""
    You can execute business tasks described in natural language by
    translating them into appropriate API calls.
    """
)

# Example usage
await automation_agent.execute_task(
    "Find all customers who haven't placed an order in the last 60 days, "
    "send them a personalized re-engagement email with a 20% discount code, "
    "and create a follow-up task for the sales team"
)

# The agent will automatically:
# 1. Call customer API to find inactive customers
# 2. Generate personalized discount codes
# 3. Call email API to send campaigns
# 4. Call task management API to create follow-up tasks

# Direct natural language to API translation
api_calls = await nl_api.translate_to_apis(
    instruction="Get the top 10 performing products from last month and update their featured status",
    context={
        "available_apis": ["products", "analytics", "inventory"],
        "user_permissions": ["read_products", "write_products", "read_analytics"]
    }
)

# Execute the generated API calls
for api_call in api_calls:
    result = await api_tool.execute(api_call)
    print(f"Executed {api_call.endpoint}: {result.status}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

from aimatrix.tools.automation import VisionAutomationTool

# Initialize vision-based automation
vision = VisionAutomationTool(
    model="gpt-4-vision",
    confidence_threshold=0.85,
    screen_resolution="auto"
)

# Handle complex enterprise applications
async def automate_erp_data_entry(data_records: list):
    """Automate data entry in legacy ERP system"""
    
    # Take screenshot to understand current state
    screenshot = await vision.capture_screen()
    
    # AI analyzes the screen
    screen_analysis = await vision.analyze_screen(
        screenshot,
        task="Identify ERP application windows and data entry forms"
    )
    
    if not screen_analysis.erp_window_found:
        # Launch ERP application
        await vision.click_on_description("ERP application icon")
        await vision.wait_for_condition("ERP login window appears")
        
        # Login
        username_field = await vision.find_element("username input field")
        await vision.type_in_element(username_field, "erp_user")
        
        password_field = await vision.find_element("password input field")
        await vision.type_in_element(password_field, "secure_password")
        
        login_button = await vision.find_element("login button")
        await vision.click_element(login_button)
        
        # Wait for main interface
        await vision.wait_for_condition("ERP main dashboard is visible")
    
    # Navigate to data entry module
    await vision.click_on_description("Data Entry module or menu item")
    await vision.wait_for_condition("Data entry form is loaded")
    
    # Process each record
    for record in data_records:
        # Start new entry
        new_record_button = await vision.find_element("New Record or Add button")
        await vision.click_element(new_record_button)
        
        # AI maps data fields to form fields
        form_mapping = await vision.map_data_to_form(
            data=record,
            form_screenshot=await vision.capture_screen()
        )
        
        # Fill form fields
        for field_name, field_value in record.items():
            if field_name in form_mapping:
                form_field = form_mapping[field_name]
                await vision.fill_field(form_field, field_value)
        
        # Save record
        save_button = await vision.find_element("Save or Submit button")
        await vision.click_element(save_button)
        
        # Wait for save confirmation
        await vision.wait_for_condition("Record saved successfully")
        
        # Handle any error dialogs
        error_dialog = await vision.find_element(
            "Error dialog or warning message",
            required=False
        )
        
        if error_dialog:
            error_text = await vision.read_text_from_element(error_dialog)
            logger.warning(f"ERP error for record {record['id']}: {error_text}")
            
            # Dismiss error dialog
            ok_button = await vision.find_element("OK or Close button in error dialog")
            await vision.click_element(ok_button)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

from aimatrix.tools.automation import MultiSystemOrchestrator

# Initialize multi-system orchestrator
orchestrator = MultiSystemOrchestrator(
    systems={
        "crm": {"type": "salesforce", "credentials": salesforce_creds},
        "erp": {"type": "sap", "credentials": sap_creds},
        "email": {"type": "outlook", "credentials": outlook_creds},
        "hr": {"type": "workday", "credentials": workday_creds}
    }
)

# Complex cross-system workflow
@orchestrator.cross_system_workflow("employee_onboarding")
async def onboard_new_employee(employee_data: dict):
    """Complete employee onboarding across all systems"""
    
    # Step 1: Create employee in HR system
    hr_result = await orchestrator.execute_on_system(
        system="hr",
        action="create_employee",
        data=employee_data
    )
    
    employee_id = hr_result["employee_id"]
    
    # Step 2: Set up email account
    email_result = await orchestrator.execute_on_system(
        system="email",
        action="create_mailbox",
        data={
            "email": f"{employee_data['first_name'].lower()}.{employee_data['last_name'].lower()}@company.com",
            "display_name": f"{employee_data['first_name']} {employee_data['last_name']}",
            "department": employee_data["department"]
        }
    )
    
    # Step 3: Add to CRM as internal user
    crm_result = await orchestrator.execute_on_system(
        system="crm",
        action="create_user",
        data={
            "employee_id": employee_id,
            "email": email_result["email"],
            "role": employee_data["role"],
            "territory": employee_data.get("territory")
        }
    )
    
    # Step 4: Create ERP access
    erp_result = await orchestrator.execute_on_system(
        system="erp",
        action="create_user_account",
        data={
            "employee_id": employee_id,
            "cost_center": employee_data["cost_center"],
            "access_level": employee_data["erp_access_level"]
        }
    )
    
    # Step 5: Send welcome package
    welcome_email = await orchestrator.execute_on_system(
        system="email",
        action="send_template_email",
        data={
            "to": email_result["email"],
            "template": "employee_welcome",
            "variables": {
                "name": employee_data["first_name"],
                "start_date": employee_data["start_date"],
                "manager": employee_data["manager"],
                "hr_contact": "hr@company.com"
            }
        }
    )
    
    # Step 6: Create onboarding tasks
    tasks = [
        "Complete I-9 form",
        "Set up direct deposit",
        "Attend orientation meeting",
        "Complete security training"
    ]
    
    for task in tasks:
        await orchestrator.execute_on_system(
            system="hr",
            action="create_task",
            data={
                "employee_id": employee_id,
                "task": task,
                "due_date": (datetime.now() + timedelta(days=7)).isoformat()
            }
        )
    
    return {
        "employee_id": employee_id,
        "email": email_result["email"],
        "crm_user_id": crm_result["user_id"],
        "erp_user_id": erp_result["user_id"],
        "status": "onboarding_initiated"
    }

# Error handling and rollback
@orchestrator.error_handler
async def handle_onboarding_error(workflow_id: str, error: Exception, completed_steps: list):
    """Handle errors in onboarding workflow"""
    
    logger.error(f"Onboarding workflow {workflow_id} failed: {str(error)}")
    
    # Rollback completed steps
    for step in reversed(completed_steps):
        try:
            if step["system"] == "hr" and step["action"] == "create_employee":
                await orchestrator.execute_on_system(
                    system="hr",
                    action="delete_employee",
                    data={"employee_id": step["result"]["employee_id"]}
                )
            elif step["system"] == "email" and step["action"] == "create_mailbox":
                await orchestrator.execute_on_system(
                    system="email",
                    action="delete_mailbox",
                    data={"email": step["result"]["email"]}
                )
            # Add other rollback actions as needed
        except Exception as rollback_error:
            logger.error(f"Rollback failed for step {step}: {str(rollback_error)}")
    
    # Notify administrators
    await orchestrator.send_alert(
        channel="#it-alerts",
        message=f"Employee onboarding workflow failed and rollback completed for workflow {workflow_id}"
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

# Robust error handling in automation
@automation_tool.with_error_handling(
    retry_attempts=3,
    retry_delay=5,
    fallback_strategy="manual_intervention"
)
async def critical_data_transfer(source_data: dict):
    try:
        # Primary automation method
        result = await primary_transfer_method(source_data)
        return result
    
    except NetworkError:
        # Try alternative network path
        result = await alternative_transfer_method(source_data)
        return result
    
    except AuthenticationError:
        # Refresh credentials and retry
        await refresh_authentication()
        result = await primary_transfer_method(source_data)
        return result
    
    except DataFormatError:
        # Transform data and retry
        transformed_data = await transform_data_format(source_data)
        result = await primary_transfer_method(transformed_data)
        return result
    
    except CriticalSystemError:
        # Escalate to human intervention
        await create_manual_intervention_ticket({
            "task": "critical_data_transfer",
            "data": source_data,
            "error": str(error),
            "priority": "high"
        })
        raise

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# Optimize automation performance
@automation_tool.performance_optimized(
    parallel_execution=True,
    caching_enabled=True,
    resource_pooling=True
)
async def batch_process_records(records: list):
    # Process records in parallel batches
    batch_size = 10
    batches = [records[i:i + batch_size] for i in range(0, len(records), batch_size)]
    
    results = []
    async with automation_tool.resource_pool(max_concurrent=5) as pool:
        for batch in batches:
            batch_tasks = [
                process_single_record(record) 
                for record in batch
            ]
            batch_results = await asyncio.gather(*batch_tasks)
            results.extend(batch_results)
    
    return results

# Smart caching for repeated operations
@automation_tool.cache(ttl=3600)  # Cache for 1 hour
async def get_system_configuration(system_name: str):
    """Cache system configuration to avoid repeated API calls"""
    config = await api_tool.get(f"https://api.{system_name}.com/config")
    return config.json()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# Secure automation with credential management
from aimatrix.security import SecureCredentialManager

credential_manager = SecureCredentialManager(
    backend="azure_key_vault",  # or "aws_secrets", "google_secret_manager"
    encryption_key="your-master-key"
)

# Never hardcode credentials
async def secure_api_call(endpoint: str, action: str):
    # Retrieve credentials securely
    credentials = await credential_manager.get_credentials(
        service=endpoint,
        user_context="automation_agent"
    )
    
    # Use temporary tokens when possible
    access_token = await get_temporary_access_token(credentials)
    
    # Make API call with secure headers
    result = await api_tool.request(
        method="POST",
        url=endpoint,
        headers={
            "Authorization": f"Bearer {access_token}",
            "User-Agent": "AIMatrix-Automation/1.0"
        },
        data=action
    )
    
    # Clear sensitive data from memory
    del access_token
    del credentials
    
    return result

1
2
3
4
5
6
7
8

# Install automation tools
pip install aimatrix[automation]

# Install browser dependencies
aimatrix setup browsers

# Configure automation environment
aimatrix configure automation --security-level enterprise

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from aimatrix import Agent
from aimatrix.tools.automation import *

# Create automation agent
automation_agent = Agent(
    name="automation-assistant",
    tools=[
        PlaywrightTool(),
        ComputerVisionTool(),
        APIOrchestrationTool(),
        WorkflowBuilder()
    ],
    instructions="""
    You are an automation expert. You can:
    1. Automate web browsers and desktop applications
    2. Orchestrate API workflows
    3. Build complex business process automation
    4. Handle errors and edge cases gracefully
    """
)

# Deploy and start automating
await automation_agent.deploy(environment="production")