Autonomous

Workflow Type: Autonomous AI Workflow

Autonomous workflows run automatically on a schedule without any manual steps. They are ideal for monitoring, automation, and periodic tasks.

Here is how an Autonomous workflow looks:

This example AI workflow uses a user-level connector (Gmail), so each user must first authorize access to their Gmail account. Once a user authorizes the connector, they can opt in to use the AI workflow for themselves. AI-Workflows are user‑scoped: each user’s runs and conversations are private to them and are not visible to other users.

You can authorize the Gmail connector either:

• By clicking the Gmail button on the AI workflow, or
• By navigating to your first-party connectors from your profile page.

For example, running this method:

auth = client.get_connector_auth(service="GMAILUSER").auth

will result in an error if you have not authorized the first-party application. Once authorized, this method can be executed from any IDE using your personal API key, and it will return an access token.

Each user has a control panel where they can start or stop the AI workflow. The Start/Stop button only affects that specific user’s subscription to the workflow. Users can opt in or opt out at any time.

On the left, users can see the historical runs of their AI workflow (for example, this workflow runs every 5 minutes). Revisiting the AI workflow allows them to review the conversations and outputs generated by these runs.

Use Case: Email Checker and Summarizer

This workflow checks Gmail every 5 minutes, fetches the latest email, and generates a summary. This is done on a per user level. This means that you can have multiple users at the same time using the AI workflow, and each one of them would have access to only the data they provided through their connectors.

---

Step 1: Define the Trigger Node

Autonomous workflows require a Loop Node that controls execution frequency.

from abacusai import WorkflowGraphNode

# Create a loop node that triggers every 5 minutes (300 seconds)
email_check_trigger_node = WorkflowGraphNode.from_template(
    template_name="Loop Node",
    name="Email Check Trigger",
    configs={},
    output_schema=[],
    sleep_time=300  # Seconds between executions
)

Key Parameters:

• template_name: Use "Loop Node" for autonomous workflows
• sleep_time: Interval in seconds (300 = 5 minutes)
• The node outputs a success boolean that triggers downstream nodes

---

Step 2: Create the Email Fetcher Function

Write a self-contained function to fetch the latest email from Gmail.

from abacusai import (
    ApiClient,
    AgentInterface, 
    AgentResponse,
    WorkflowGraph,
    WorkflowGraphNode,
    WorkflowNodeInputMapping,
    WorkflowNodeInputSchema,
    WorkflowNodeInputType,
    WorkflowNodeOutputMapping,
    WorkflowNodeOutputSchema,
    WorkflowNodeOutputType,
)

def fetch_latest_email(success):
    """
    Fetches the most recent email from Gmail inbox
    
    Args:
        success (bool): Trigger from loop node
        
    Returns:
        AgentResponse: Contains email_content and email_subject
    """
    # ALL imports must be inside the function
    from abacusai import AgentResponse, ApiClient
    import requests
    import base64

    # Initialize Abacus AI client
    client = ApiClient()
    
    # Get Gmail OAuth credentials
    auth = client.get_connector_auth(service="GMAILUSER").auth
    headers = {
        "Accept": "application/json",
        "Authorization": f"Bearer {auth['accessToken']}",
    }
    
    # List the most recent email
    list_url = "https://gmail.googleapis.com/gmail/v1/users/me/messages"
    params = {"maxResults": 1}
    resp = requests.get(list_url, headers=headers, params=params)
    resp.raise_for_status()
    
    messages = resp.json().get("messages", [])
    if not messages:
        return AgentResponse(
            email_content="No emails found",
            email_subject="No emails"
        )
    
    # Get email details
    msg_id = messages[0]["id"]
    detail_url = f"https://gmail.googleapis.com/gmail/v1/users/me/messages/{msg_id}"
    detail_resp = requests.get(detail_url, headers=headers, params={"format": "full"})
    detail_resp.raise_for_status()
    email_data = detail_resp.json()
    
    # Extract subject
    subject = "No Subject"
    headers_list = email_data.get("payload", {}).get("headers", [])
    for header in headers_list:
        if header.get("name") == "Subject":
            subject = header.get("value", "No Subject")
            break
    
    # Extract body
    body = ""
    payload = email_data.get("payload", {})
    
    if "parts" in payload:
        for part in payload["parts"]:
            if (part.get("mimeType") == "text/plain") and part.get("body", {}).get("data"):
                try:
                    data_str = part["body"]["data"]
                    body = base64.urlsafe_b64decode(data_str.encode("utf-8")).decode(
                        "utf-8", errors="ignore"
                    )
                    break
                except Exception:
                    continue
    elif payload.get("body", {}).get("data"):
        try:
            data_str = payload["body"]["data"]
            body = base64.urlsafe_b64decode(data_str.encode("utf-8")).decode(
                "utf-8", errors="ignore"
            )
        except Exception:
            body = email_data.get("snippet", "")
    
    if not body:
        body = email_data.get("snippet", "")
    
    return AgentResponse(email_content=body, email_subject=subject)

---

Step 3: Transform Function to Node

Convert the Python function into a WorkflowGraphNode with input/output mappings.

fetch_latest_email_node = WorkflowGraphNode(
    "Fetch Latest Email",                    # Node name (displayed in UI)
    fetch_latest_email,                       # The Python function
    input_mappings={
        "success": email_check_trigger_node.outputs.success  # Connect to trigger
    },
    output_mappings=["email_content", "email_subject"]  # Outputs from AgentResponse
)

Understanding Mappings:

• input_mappings: Dictionary connecting upstream outputs to this function's parameters
• output_mappings: List of fields from the AgentResponse object
• Data flows: trigger.success → fetch_latest_email(success)

---

Step 4: Create the Email Summarizer Function

def summarize_email(email_content, email_subject):
    """
    Summarizes the email content using LLM
    
    Args:
        email_content (str): Body of the email
        email_subject (str): Subject line
        
    Returns:
        AgentResponse: Contains summary text
    """
    from abacusai import AgentResponse, ApiClient

    client = ApiClient()
    
    # Handle empty emails
    if (not email_content) or (email_content.strip() == ""):
        return AgentResponse(summary="No email content to summarize")
    
    # Create prompt for LLM
    prompt = f"""
    Email Subject: {email_subject}
    Email Content: {email_content}

    Please provide a concise summary of this email.
    """
    
    system_message = """You are an expert email summarizer. 
    Create a brief, clear summary that captures the key points and action items 
    from the email. Keep it under 100 words."""
    
    # Call LLM to generate summary
    summary_response = client.evaluate_prompt(
        prompt=prompt,
        system_message=system_message,
        llm_name="GEMINI_2_FLASH"  # Fast, efficient model
    )
    
    return AgentResponse(summary=summary_response.content)

---

Step 5: Create Summarizer Node

summarize_email_node = WorkflowGraphNode(
    "Summarize Email",
    summarize_email,
    input_mappings={
        # Connect both outputs from fetch_latest_email_node
        "email_content": fetch_latest_email_node.outputs.email_content,
        "email_subject": fetch_latest_email_node.outputs.email_subject,
    },
    output_mappings=["summary"]
)

Data Flow:

Trigger → Fetch Email → Summarize Email
         (content, subject)   → (summary)

---

Step 6: Build the Workflow Graph

# Define workflow as autonomous
agent_interface = AgentInterface.AUTONOMOUS

# Create the workflow graph
workflow_graph = WorkflowGraph(
    nodes=[
        email_check_trigger_node,
        fetch_latest_email_node,
        summarize_email_node
    ],
    specification_type="data_flow"  # Nodes execute based on data dependencies
)

---

Step 7: Register the Workflow as an Agent

Option A: Update Existing Agent

from abacusai import ApiClient
client = ApiClient()

agent = client.update_agent(
    model_id='your_model_id',
    workflow_graph=workflow_graph,
    agent_interface=agent_interface,
    description="Autonomous email checker that fetches and summarizes emails every 5 minutes",
    package_requirements=[],                
    org_level_connectors=[],
    user_level_connectors={'GMAILUSER': ['https://www.googleapis.com/auth/gmail.modify']},
    included_modules=[],
)

# Wait for deployment to complete
agent.wait_for_publish()
print(f"Agent updated successfully: {agent.agent_id}")

Option B: Create New Agent

agent = client.create_agent(
    project_id='your_project_id',          # Your Abacus AI project
    workflow_graph=workflow_graph,
    agent_interface=agent_interface,
    description="Autonomous email checker that fetches and summarizes emails every 5 minutes",
    memory=True,
    package_requirements=[],
    org_level_connectors=[],
    user_level_connectors={'GMAILUSER': ['https://www.googleapis.com/auth/gmail.modify']},    
    included_modules=[],
)

agent.wait_for_publish()
print(f"Agent created successfully: {agent.agent_id}")

Key Parameters:

• agent_interface: Must be AgentInterface.AUTONOMOUS for autonomous workflows
• user_level_connectors: Dictionary of required external services (Gmail, Slack, etc.)
• package_requirements: List of additional pip packages (e.g., ["pandas", "numpy"])