Tasks | m8tes

A task is saved work attached to an agent. Use tasks when you want repeatable instructions that can run on demand or by trigger.

Create a task

from m8tes import M8tes

client = M8tes()

task = client.tasks.create(
    agent_id=1,
    name="daily support review",
    instructions="review unread support emails, rank priority, and draft replies",
)

curl -X POST https://api.m8tes.ai/api/v2/tasks \
  -H "Authorization: Bearer m8_your_key_here" \
  -H "Content-Type: application/json" \
  -d '{"teammate_id": 1, "name": "daily support review", "instructions": "review unread support emails, rank priority, and draft replies"}'

teammate_id is the wire name for the agent id: JSON request and response bodies keep it. The SDK accepts agent_id (canonical) and teammate_id alike.

Key fields

Field	Type	Description
`teammate_id`	int (required)	Agent that executes the task
`instructions`	string (required)	What should happen when the task runs
`name`	string	Task label
`schedule`	string	Cron expression, e.g. `0 9 * * 1-5`. Attaches trigger immediately
`webhook`	bool	Enable webhook trigger at creation. Returns `webhook_url` once
`user_id`	string	End-user scope. See Users
`workflow_stage`	string	Optional board position: `unassigned`, `pending`, `planning`, `queue`, `review`, or `done`

All fields

Field	Type	Description
`tools`	string[]	Optional override of agent tools
`expected_output`	string	Output format expectation
`goals`	string	Task-specific goals injected into prompt context
`schedule_timezone`	string	Timezone for `schedule`. Defaults to `UTC`
`enable_lessons`	bool	Accumulate lessons from runs. Defaults to `true`. See Lessons

webhook_enabled is true when a webhook trigger is active. webhook_url is only set on first creation when webhook=True; it's not retrievable afterwards.

Run a task

Python

with client.tasks.run(task.id) as stream:
    for event in stream:
        if event.type == "text-delta":
            print(event.delta, end="", flush=True)

Approval callbacks, non-streaming runs, tool scoping

run_and_wait() handles approval pauses inline, no polling loop needed. It accepts the same callbacks as runs.create_and_wait(); see Human-in-the-Loop.

Python

run = client.tasks.run_and_wait(
    task.id,
    on_approval=lambda req: "allow",
    on_question=lambda req: {"Which channel?": "#general"},
)
print(run.output)

Non-streaming:

Python

run = client.tasks.run(task.id, stream=False)
run = client.runs.poll(run.id)
print(run.output)

Pass task_setup_tools=False to keep a run limited to the task's normal tools, skipping the internal same-scope management tools for tasks, runs, inboxes, webhooks, and integrations. Pass feedback=False to disable the internal issue-reporting tool (report_issue).

Scheduling

Pass schedule= at creation time to attach a cron trigger in the same call:

Python

task = client.tasks.create(
    agent_id=1,
    instructions="send weekly revenue report to #finance",
    schedule="0 9 * * 1-5",                # weekdays at 9:00
    schedule_timezone="America/New_York",  # defaults to UTC
)

See Scheduling for cron patterns, interval schedules, and pausing or reshaping schedules.

List every trigger across the account without fetching each task separately:

Python

for trigger in client.triggers.list(type="schedule", user_id="customer_123").auto_paging_iter():
    print(trigger.task_id, trigger.id, trigger.enabled)

client.triggers.list() includes schedule, webhook, email, and app triggers, including disabled ones. The legacy V1 schedules list returned only enabled schedules.

Webhook trigger

Python

task = client.tasks.create(agent_id=1, instructions="process incoming data", webhook=True)
print(task.webhook_url)  # POST here to trigger a run (shown once, save it)

Add one to an existing task with client.tasks.triggers.create(task.id, type="webhook"); the returned trigger.url is likewise shown once.

Retrieve a task's results

Every run reports its task (run.task_id) and runs.list filters by it, so a scheduled or webhook-triggered task's history and outputs are one call away:

Python

latest = client.runs.list(task_id=task.id, user_id="customer_123", status="completed", limit=1).data
if latest:
    print(client.runs.get(latest[0].id).output)

Failed runs carry a machine-readable error_code plus a human-readable output; retryable=True means client.runs.retry(run.id) will be accepted. Prefer push? Subscribe a webhook endpoint to run.completed / run.failed.

Lessons

Agents save lessons from a task's runs: durable corrections and preferences they apply on future runs, accumulated automatically (capped at 20 per task; disable with enable_lessons=False). The API is the curation surface: read what the agent learned, remove bad entries.

Python

for lesson in client.tasks.lessons(task.id).data:
    print(lesson.id, lesson.text, lesson.when_applicable)
client.tasks.delete_lesson(task.id, lesson_id="...")   # or clear_lessons(task.id) for all

Manage tasks

Python

for item in client.tasks.list(agent_id=1, user_id="customer_123").data:
    print(item.id, item.name)
client.tasks.update(task.id, instructions="review unread emails, escalate urgent incidents, and draft replies")
client.tasks.delete(task.id)

Next: Runs · Scheduling · Human-in-the-Loop · Webhook Events