Polling for Results

1
import time
2
import requests
3
4
def poll_execution(
5
    workflow_id: str,
6
    execution_id: str,
7
    access_token: str,
8
    initial_interval: int = 5,
9
    max_interval: int = 30,
10
    max_wait: int = 600
11
) -> dict:
12
    """
13
    Poll a workflow execution until it completes or fails.
14
15
    Uses exponential backoff starting at initial_interval seconds,
16
    capping at max_interval seconds.
17
    """
18
    base_url = "https://api.further.ai"
19
    url = f"{base_url}/api/v1/workflow-builder/workflows/{workflow_id}/executions/{execution_id}"
20
    headers = {"Authorization": f"Bearer {access_token}"}
21
22
    elapsed = 0
23
    interval = initial_interval
24
25
    while elapsed < max_wait:
26
        response = requests.get(url, headers=headers)
27
        response.raise_for_status()
28
        result = response.json()
29
30
        status = result["status"]
31
        print(f"[{elapsed}s] Status: {status}")
32
33
        if status == "completed":
34
            return result
35
        if status == "failed":
36
            return result
37
38
        time.sleep(interval)
39
        elapsed += interval
40
        interval = min(interval * 1.5, max_interval)
41
42
    raise TimeoutError(f"Execution did not complete within {max_wait} seconds")
43
44
45
# Usage
46
result = poll_execution(
47
    workflow_id="YOUR_WORKFLOW_ID",
48
    execution_id="YOUR_EXECUTION_ID",
49
    access_token=access_token
50
)
51
52
if result["status"] == "completed":
53
    for step in result["steps"]:
54
        print(f"{step['step_name']}: {step['data']}")