Skip to main content

Tasks

Tasks are user-owned work items, optionally recurring and scoped to an agent/app/workflow. Use this resource for CRUD plus lifecycle (complete, cancel, skip, end recurrence).

Access it as client.tasks on a Strongly client, or the same path on AsyncStrongly with await. All methods exist on both with identical signatures.

Quick start​

from strongly import Strongly

client = Strongly()

# List (auto-paginates as you iterate)  # filters: kind, status, include_completed, search_text, agent_id, app_id, workflow_id
for task in client.tasks.list():
    print(task.id)

# Create
created = client.tasks.create(
    description="...",
)
print(created.id)

# Read it back
fetched = client.tasks.retrieve(created.id)

# Delete (returns None; raises on failure)
client.tasks.delete(created.id)

Methods​

Core​

list​

list(*, kind: Optional[str] = None, status: Optional[str] = None, include_completed: Optional[bool] = None, search_text: Optional[str] = None, agent_id: Optional[str] = None, app_id: Optional[str] = None, workflow_id: Optional[str] = None, limit: Optional[int] = None) -> SyncPaginator[Task]

List tasks with pagination and filters.

Args: kind: recurring or oneoff. status: open | fired | done | failed | cancelled. include_completed: Include done/failed/cancelled rows (default off). search_text: Substring match on description and notes. agent_id: Restrict to tasks scoped to this agent. app_id: Restrict to tasks scoped to this app instance. workflow_id: Restrict to tasks scoped to this workflow. limit: Maximum number of items to return (default: all matching items).

Returns: A paginator yielding Task objects.

create​

create(*, description: str, subject_ref: Optional[Mapping[str, Any]] = None, due_at: Optional[str] = None, notes: Optional[str] = None, recurrence: Optional[Mapping[str, Any]] = None, tags: Optional[Sequence[str]] = None, scope: Optional[Mapping[str, Any]] = None, created_by_agent: Optional[str] = None, assigned_to_agent: Optional[str] = None, shared_with: Optional[Sequence[str]] = None, is_public: Optional[bool] = None) -> Task

Create a task.

Recurring tasks (recurrence.kind != "none") get a computed next_due_at; one-off tasks use the provided due_at.

Args: description: What needs to be done (required). subject_ref: Optional {kind, id, app?} pointer to the subject. due_at: ISO timestamp for one-off tasks. notes: Free-form notes. recurrence: A recurrence dict with camelCase wire keys, e.g. {"kind": "interval", "interval": 1, "timeOfDay": "09:00"}. tags: Tag labels. scope: Scope the task to a resource, e.g. {"agentId": ..., "appId": ..., "workflowId": ...}. created_by_agent, assigned_to_agent: Authoring/assignment agent ids. shared_with: User ids to share the task with on creation. is_public: Whether the task is publicly visible.

Returns: The created Task.

retrieve​

retrieve(task_id: str) -> Task

Retrieve a single task by id.

update​

update(task_id: str, *, description: Optional[str] = None, subject_ref: Optional[Mapping[str, Any]] = None, due_at: Optional[str] = None, notes: Optional[str] = None, recurrence: Optional[Mapping[str, Any]] = None, tags: Optional[Sequence[str]] = None, status: Optional[str] = None, scope: Optional[Mapping[str, Any]] = None) -> Task

Update a task. Changing recurrence recomputes next_due_at.

delete​

delete(task_id: str) -> None

Delete a task (hard delete). Raises on failure; returns nothing.

Lifecycle & actions​

cancel​

cancel(task_id: str, *, reason: Optional[str] = None) -> Task

Cancel a task (sets status cancelled). Optional reason.

complete​

complete(task_id: str, *, notes: Optional[str] = None, execution_id: Optional[str] = None) -> Task

Mark a task complete and return the updated task.

Recurring tasks advance to the next due instant; one-off tasks transition to done.

Args: task_id: Target task id. notes: Optional completion notes. execution_id: Workflow execution that completed the task.

skip_next​

skip_next(task_id: str) -> Task

Skip the next scheduled fire on a recurring task.

end_recurrence​

end_recurrence(task_id: str) -> Task

Stop recurrence on a recurring task (history is preserved).

Sharing & scope​

share​

share(task_id: str, user_ids: List[str]) -> Dict[str, Any]

Share a task with the given user ids.

unshare​

unshare(task_id: str, user_ids: List[str]) -> Dict[str, Any]

Revoke a task share from the given user ids.

toggle_public​

toggle_public(task_id: str) -> Dict[str, Any]

Toggle the public visibility flag on a task.

update_scope​

update_scope(task_id: str, *, scope: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Re-scope a task (owner-only). An empty scope moves it to user-global.

Args: task_id: Target task id. scope: {agentId?, appId?, workflowId?}.

Returns: {success, scope} with the resolved scope.

Other​

status​

status(task_id: str) -> TaskStatus

Retrieve the lightweight status/scheduling block for a task.