Long-Running AOP Execution | Athena

This guide shows how to execute Agent Operating Procedures (AOPs) using the Python SDK with asynchronous execution and proper polling for completion. This is the recommended approach for production applications where AOPs may take minutes or longer to complete.

Why Async Execution? When you call execute_async(), it returns immediately with a thread_id. The AOP continues running in the background. You must poll threads.get_status() in a loop to know when execution finishes. Without polling, you have no way to know if the AOP completed, failed, or is still running.

Key features:

Non-blocking execution - execute_async() returns immediately with a thread ID
Polling-based completion tracking - Use threads.get_status() in a loop to wait for results
Configurable timeouts - Set appropriate timeouts for your workflow complexity
Production-ready patterns - Error handling, retries, and batch processing

execute_async() returns immediately with a thread_id and does not wait for completion. You must implement a polling loop using threads.get_status() to wait for the AOP to finish.

Install Package

!pip install -U athenaintel

Set Up Client

import os
import time
from athena.client import Athena
client = Athena(api_key=os.environ["ATHENA_API_KEY"])

Start Async Execution

Call execute_async() to start the AOP. This returns immediately with a thread_id you use to track progress.

response = client.aop.execute_async(
    asset_id="asset_xxx",
    user_inputs={"company": "Acme Corp"}
)
thread_id = response.thread_id
print(f"Started AOP: {response.aop_title}")
print(f"Thread ID: {thread_id}")
print(f"Status: {response.status}")

At this point the AOP is running in the background. The response.status will indicate it has started, but the work is not done yet.

Poll for Completion

You must check threads.get_status() in a loop until the status is completed or failed.

timeout = 3600
poll_interval = 5
start_time = time.time()
while True:
    if time.time() - start_time > timeout:
        raise TimeoutError(f"AOP execution timed out after {timeout}s")
    status = client.threads.get_status(thread_id=thread_id)
    if status.status == "completed":
        print("AOP completed successfully!")
        break
    elif status.status == "failed":
        print(f"AOP execution failed")
        break
    time.sleep(poll_interval)
print(f"Final status: {status.status}")

Recommended Configuration

Parameter	Recommended Value	Notes
Poll interval	5 seconds	Balance between responsiveness and API load
Timeout	3600 seconds (1 hour)	Increase for complex AOPs, decrease for simple ones
Statuses to handle	`running`, `completed`, `failed`, `waiting-for-approval`	Always handle terminal states

Production-Ready Helper Function

Wrap the full pattern into a reusable function:

import time
from athena.client import Athena
from athena.core import ApiError
def execute_aop_with_polling(
    client: Athena,
    aop_asset_id: str,
    user_inputs: dict,
    timeout_seconds: int = 3600,
    poll_interval: int = 5
):
    """Execute an AOP and poll until completion."""
    response = client.aop.execute_async(
        asset_id=aop_asset_id,
        user_inputs=user_inputs
    )
    thread_id = response.thread_id
    start_time = time.time()
    while True:
        elapsed = time.time() - start_time
        if elapsed > timeout_seconds:
            raise TimeoutError(
                f"AOP execution timed out after {elapsed:.1f}s"
            )
        status_response = client.threads.get_status(
            thread_id=thread_id
        )
        if status_response.status == "completed":
            return status_response
        elif status_response.status == "failed":
            raise RuntimeError(
                f"AOP execution failed for thread {thread_id}"
            )
        time.sleep(poll_interval)

Usage:

client = Athena(api_key=os.environ["ATHENA_API_KEY"])
result = execute_aop_with_polling(
    client=client,
    aop_asset_id="asset_xxx",
    user_inputs={"company": "Acme Corp"},
    timeout_seconds=1800,
    poll_interval=5
)
print(f"Completed: {result.status}")

Batch Processing

Process multiple AOPs sequentially with proper polling for each:

def execute_aops_sequentially(
    client: Athena,
    aop_configs: list[dict],
    timeout_seconds: int = 3600,
    poll_interval: int = 5
) -> list[dict]:
    """Execute multiple AOPs one after another, polling each to completion."""
    results = []
    for i, config in enumerate(aop_configs):
        print(f"[{i + 1}/{len(aop_configs)}] Starting: {config['asset_id']}")
        try:
            result = execute_aop_with_polling(
                client=client,
                aop_asset_id=config["asset_id"],
                user_inputs=config.get("user_inputs", {}),
                timeout_seconds=timeout_seconds,
                poll_interval=poll_interval
            )
            results.append({
                "asset_id": config["asset_id"],
                "status": "completed",
                "result": result
            })
        except (TimeoutError, RuntimeError) as e:
            results.append({
                "asset_id": config["asset_id"],
                "status": "failed",
                "error": str(e)
            })
    return results
# Usage
aop_configs = [
    {"asset_id": "asset_research_aop", "user_inputs": {"company": "Acme Corp"}},
    {"asset_id": "asset_analysis_aop", "user_inputs": {"quarter": "Q1 2024"}},
    {"asset_id": "asset_report_aop", "user_inputs": {"format": "summary"}},
]
results = execute_aops_sequentially(client, aop_configs)
for r in results:
    print(f"{r['asset_id']}: {r['status']}")

Error Handling

Handle the full range of errors that can occur during async execution:

import time
from athena.client import Athena
from athena.core import ApiError
def execute_aop_robust(
    client: Athena,
    aop_asset_id: str,
    user_inputs: dict,
    timeout_seconds: int = 3600,
    poll_interval: int = 5,
    max_poll_errors: int = 3
):
    """Execute an AOP with comprehensive error handling."""
    # Start execution
    try:
        response = client.aop.execute_async(
            asset_id=aop_asset_id,
            user_inputs=user_inputs
        )
    except ApiError as e:
        raise RuntimeError(
            f"Failed to start AOP: {e.status_code} - {e.body}"
        )
    thread_id = response.thread_id
    start_time = time.time()
    consecutive_errors = 0
    while True:
        elapsed = time.time() - start_time
        if elapsed > timeout_seconds:
            raise TimeoutError(
                f"AOP timed out after {elapsed:.1f}s. "
                f"Thread {thread_id} may still be running."
            )
        try:
            status = client.threads.get_status(thread_id=thread_id)
            consecutive_errors = 0
        except ApiError:
            consecutive_errors += 1
            if consecutive_errors >= max_poll_errors:
                raise RuntimeError(
                    f"Lost connection: {consecutive_errors} consecutive "
                    f"polling failures for thread {thread_id}"
                )
            time.sleep(poll_interval)
            continue
        if status.status == "completed":
            return status
        elif status.status == "failed":
            raise RuntimeError(
                f"AOP execution failed for thread {thread_id}"
            )
        elif status.status == "waiting-for-approval":
            print(
                f"Thread {thread_id} is waiting for approval. "
                f"Approve in the Athena UI to continue."
            )
        time.sleep(poll_interval)

Handled error scenarios:

Error	Cause	Handling
`TimeoutError`	AOP takes longer than `timeout_seconds`	Raise with thread ID so you can check later
`ApiError` on start	Invalid asset ID, auth failure, bad inputs	Raise immediately, do not poll
`ApiError` during poll	Transient network issue	Retry up to `max_poll_errors` times, then raise
`failed` status	AOP execution error	Raise `RuntimeError` with thread ID
`waiting-for-approval`	AOP requires human approval step	Log a message, keep polling

Common Mistakes

Single status check (wrong):

# This only checks ONCE and does not wait for completion
response = client.aop.execute_async(
    asset_id="asset_xxx",
    user_inputs={"company": "Acme Corp"}
)
status = client.threads.get_status(thread_id=response.thread_id)
# BUG: AOP is still running, status is NOT "completed" yet

Polling loop (correct):

# This properly waits for completion
response = client.aop.execute_async(
    asset_id="asset_xxx",
    user_inputs={"company": "Acme Corp"}
)
while True:
    status = client.threads.get_status(thread_id=response.thread_id)
    if status.status in ["completed", "failed"]:
        break
    time.sleep(5)

The difference: execute_async() starts execution and returns immediately. If you only check status once, the AOP is almost certainly still running. You need the while True loop to keep checking until a terminal status is reached.