Skip to content

list_tasks

autogen.beta.a2a.tasks.list_tasks async #

list_tasks(config, *, tenant=None, context_id=None, status=None, page_size=None, page_token=None, history_length=None, include_artifacts=False, status_timestamp_after=None)

List tasks; pagination is the caller's responsibility via next_page_token.

Returns a :class:ListedTasks carrying both the page's tasks and the server-reported pagination metadata. Iterate over the result or call .tasks for the list directly.

Source code in autogen/beta/a2a/tasks.py
async def list_tasks(
    config: A2AConfig,
    *,
    tenant: str | None = None,
    context_id: str | None = None,
    status: TaskState | None = None,
    page_size: int | None = None,
    page_token: str | None = None,
    history_length: int | None = None,
    include_artifacts: bool = False,
    status_timestamp_after: datetime | None = None,
) -> ListedTasks:
    """List tasks; pagination is the caller's responsibility via ``next_page_token``.

    Returns a :class:`ListedTasks` carrying both the page's tasks and the
    server-reported pagination metadata. Iterate over the result or call
    ``.tasks`` for the list directly.
    """
    async with open_session(config) as sdk:
        kwargs = with_tenant(config, tenant)
        optional = {
            "context_id": context_id,
            "status": status,
            "page_size": page_size,
            "page_token": page_token,
            "history_length": _resolve_history(config, history_length),
            "status_timestamp_after": status_timestamp_after,
        }
        kwargs.update({k: v for k, v in optional.items() if v is not None})
        if include_artifacts:
            kwargs["include_artifacts"] = True
        response = await sdk.list_tasks(ListTasksRequest(**kwargs))
        return ListedTasks(
            tasks=list(response.tasks),
            next_page_token=response.next_page_token,
            page_size=response.page_size,
            total_size=response.total_size,
        )