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,
)
|