AgentRailClient task lifecycle methods — Python SDK

The Python AgentRailClient provides async methods for every stage of the AgentRail task lifecycle. All method names use snake_case and all response data is returned as validated Pydantic v2 models. You construct the client once and use it across your harness, typically as an async context manager.

Construct the client

import asyncio
import os

from agentrail import AgentRailClient, TaskStatus

async def main():
    api_key = os.environ["AGENTRAIL_API_KEY"]
    base_url = os.getenv("AGENTRAIL_BASE_URL", "http://127.0.0.1:3000")

    async with AgentRailClient(base_url=base_url, api_key=api_key) as client:
        tasks = await client.list_my_tasks(status=TaskStatus.TODO, limit=10)

        for task in tasks.data:
            print(f"{task.identifier}: {task.title}")
            print(f"Next actions: {', '.join(task.available_actions)}")

asyncio.run(main())

Python models expose snake_case attributes. The available_actions list tells you what operations are currently permitted on each task.

Submit, observe, and ship

The following example shows the complete lifecycle: submitting work, reading CI and review status, then shipping when all conditions are met.

import asyncio
import os

from agentrail import (
    AgentRailClient,
    PullRequestSubmitOptions,
    ShipEnvironment,
    ShipMode,
    TaskShipRequest,
    TaskStatus,
    TaskSubmitMode,
    TaskSubmitRequest,
)


async def main():
    async with AgentRailClient(
        base_url=os.getenv("AGENTRAIL_BASE_URL", "http://127.0.0.1:3000"),
        api_key=os.environ["AGENTRAIL_API_KEY"],
    ) as client:
        tasks = await client.list_my_tasks(status=TaskStatus.IN_PROGRESS, limit=1)
        if not tasks.data:
            return

        task = tasks.data[0]
        detail = await client.get_task(task.id)

        if "submit" in detail.data.available_actions:
            submission = await client.submit_task(
                task.id,
                TaskSubmitRequest(
                    summary="Implemented the task and pushed commits to the task branch.",
                    mode=TaskSubmitMode.ADAPTER_MANAGED,
                    pull_request=PullRequestSubmitOptions(
                        title=f"Submit {detail.data.identifier}",
                        draft=False,
                    ),
                ),
                idempotency_key=f"submit-{task.id}-v1",
            )
            print(f"PR: {submission.data.pr_url}")

        ci = await client.get_task_ci_status(task.id)
        review = await client.get_task_review_feedback(task.id)
        refreshed = await client.get_task(task.id)

        head_sha = (
            ci.data.head_sha
            or review.data.latest_decision.head_sha
            or refreshed.data.head_sha
        )

        if (
            "ship" in refreshed.data.available_actions
            and ci.data.overall_status.value == "passed"
            and review.data.latest_decision.outcome.value == "approved"
            and head_sha
        ):
            await client.ship_task(
                task.id,
                TaskShipRequest(
                    mode=ShipMode.MERGE_AND_DEPLOY,
                    target_environment=ShipEnvironment.PRODUCTION,
                    expected_head_sha=head_sha,
                ),
                idempotency_key=f"ship-{task.id}-{head_sha}",
            )


asyncio.run(main())

Always gate shipping on all four conditions: "ship" in available_actions, overall_status.value == "passed", outcome.value == "approved", and a resolved head_sha. Sending a stale SHA returns a 409.

Method reference

`list_my_tasks`

tasks = await client.list_my_tasks(status=TaskStatus.TODO, limit=10)

Pass a TaskStatus enum value to filter by status. Available values include TaskStatus.TODO, TaskStatus.IN_PROGRESS, and others.

`get_task`

detail = await client.get_task(task_id)

Returns a TaskDetailResponse with the full task object including available_actions, branch, head SHA, and current state.

`submit_task`

submission = await client.submit_task(
    task_id,
    TaskSubmitRequest(
        summary="Work description.",
        mode=TaskSubmitMode.ADAPTER_MANAGED,
        pull_request=PullRequestSubmitOptions(title="PR title", draft=False),
    ),
    idempotency_key="submit-tsk_123-v1",
)

Use TaskSubmitMode.ADAPTER_MANAGED for real provider automation. AgentRail creates or reuses the provider PR and returns the PR metadata in submission.data.pr_url.

`get_task_ci_status`

ci = await client.get_task_ci_status(task_id)
# ci.data.overall_status.value: "passed" | "failed" | "pending" | ...
# ci.data.head_sha: str | None

`get_task_review_feedback`

review = await client.get_task_review_feedback(task_id)
# review.data.latest_decision.outcome.value: "approved" | "changes_requested" | ...
# review.data.latest_decision.head_sha: str | None

`ship_task`

await client.ship_task(
    task_id,
    TaskShipRequest(
        mode=ShipMode.MERGE_AND_DEPLOY,
        target_environment=ShipEnvironment.PRODUCTION,
        expected_head_sha=head_sha,
    ),
    idempotency_key=f"ship-{task_id}-{head_sha}",
)

expected_head_sha acts as an optimistic lock. The ship is rejected if the branch has advanced since you read the SHA.

Idempotency keys

submit_task and ship_task both require an idempotency_key keyword argument. See authentication for key naming conventions.

Overview

TypeScript

Python

AgentRailClient task lifecycle methods — Python SDK

Construct the client

Submit, observe, and ship

Method reference

`list_my_tasks`

`get_task`

`submit_task`

`get_task_ci_status`

`get_task_review_feedback`

`ship_task`

Idempotency keys

​Construct the client

​Submit, observe, and ship

​Method reference

​list_my_tasks

​get_task

​submit_task

​get_task_ci_status

​get_task_review_feedback

​ship_task

​Idempotency keys

Construct the client

Submit, observe, and ship

Method reference

`list_my_tasks`

`get_task`

`submit_task`

`get_task_ci_status`

`get_task_review_feedback`

`ship_task`

Idempotency keys