> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.athenaintel.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.athenaintel.com/_mcp/server.

# Long-Running AOP Execution

> Execute AOPs asynchronously with proper polling for completion. Essential pattern for reliable AOP execution.

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

> **Warning**
>
> `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

#### Jupyter Notebook

```python
!pip install -U athenaintel
```

#### Terminal/Shell

```bash
pip install -U athenaintel
```

### Set Up Client

```python
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.

```python
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`.

```python
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:

```python
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:

```python
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:

```python
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:

```python
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):**

```python
# 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):**

```python
# 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.