ExecutionsResource: List, Inspect, and Stream Runs

list

list(*, api_key_route=True, options=None) -> {"executions": List[Execution]}

Returns a list of executions. By default it calls GET /api/executions/list, which is the route intended for API-key access. If you pass api_key_route=False, the SDK switches to GET /api/executions instead.

Most callers should leave api_key_route at its default (True). Only override it when your deployment requires the alternate route.

# Default route: GET /api/executions/list
result = client.executions.list()
print(result["executions"][0]["runId"], result["executions"][0]["status"])

# Alternate route: GET /api/executions
result = client.executions.list(api_key_route=False)

get

get(execution_id, *, options=None) -> {"execution": Execution}

Fetches a single execution by its ID. Calls GET /api/executions/{id}.

result = client.executions.get("exec_01JXYZ")
print(result["execution"]["status"], result["execution"]["durationMs"])

delete

delete(execution_id, *, options=None) -> {"success": bool}

Removes a single execution record. Calls DELETE /api/executions/{id}.

result = client.executions.delete("exec_01JXYZ")
print(result["success"])

clear

clear(*, options=None) -> {"success": bool}

Deletes all executions in bulk. Calls POST /api/executions/clear.

result = client.executions.clear()
print(result["success"])

stop

stop(run_id, *, options=None) -> {"success": bool}

Sends a stop signal to an active run. Calls POST /api/executions/stop.

result = client.executions.stop("run_abc123")
print(result["success"])

stream

stream(*, options=None) -> Iterator[StreamEvent]

Opens a Server-Sent Events connection to GET /api/executions/stream and yields each event as it arrives. The iterator stays open until the server closes the stream, the client stops iterating, or the request is aborted.For details on the StreamEvent shape and how cancellation works, see Streaming. For error handling, see Errors.

import figranium

try:
    for event in client.executions.stream(options={"timeout": 60}):
        print(event["event"], event["data"])
except figranium.FigraniumError as error:
    # FigraniumError with code "REQUEST_ABORTED" when timed out
    print(error)

On AsyncFigranium, stream returns an AsyncIterator[StreamEvent] and you iterate with async for.

async for event in await async_client.executions.stream():
    print(event["event"], event["data"])

JavaScript SDK

Python SDK

Swift SDK

ExecutionsResource: List, Inspect, and Stream Runs

Methods

End-to-end example

​Methods

​End-to-end example

Methods

End-to-end example