create_task - xpander.ai

Agent.acreate_task creates a new task on the platform and returns a Task object you can save, stream, or stop. This is the primary way to invoke an agent from code.

from xpander_sdk import Agents

agent = await Agents().aget("agent-123")
task = await agent.acreate_task(prompt="Summarize this document.")
print(task.id, task.status.value)

The newly-created task is queued by the platform; the agent worker (either a managed cloud worker or your local @on_task handler) picks it up and executes it.

Parameters

Parameter	Type	Required	Default	Description
`prompt`	`str`	No	`""`	Natural-language input.
`existing_task_id`	`str`	No	`None`	Continue an existing task. Reuses memory thread.
`file_urls`	`list[str]`	No	`[]`	URLs of files to attach (PDFs, images, CSVs, …).
`user_details`	`User`	No	`None`	End-user context (id, email, timezone) for memory and connector auth.
`agent_version`	`str`	No	`None`	Pin to a specific agent version.
`tool_call_payload_extension`	`dict`	No	`None`	Extra fields merged into every tool-call payload.
`source`	`str`	No	`"sdk"`	Origin tag (`"webhook"`, `"slack"`, …).
`worker_id`	`str`	No	`None`	Pin execution to a specific worker.
`run_locally`	`bool`	No	`False`	Mark the task as locally executed (skips cloud dispatch).
`output_format`	`OutputFormat`	No	`None`	`Text`, `Markdown`, `Json`, or `Voice`.
`output_schema`	`dict`	No	`None`	JSON schema (paired with `OutputFormat.Json`).
`events_streaming`	`bool`	No	`False`	Enables SSE streaming via `task.aevents()`.
`additional_context`	`str`	No	`None`	Extra context appended to the system prompt.
`instructions_override`	`str`	No	`None`	Extra instructions appended for this run only.
`test_run_node_id`	`str`	No	`None`	(Internal) Workflow node to execute in test mode.
`user_oidc_token`	`str`	No	`None`	OIDC token for connector pre-auth.
`expected_output`	`str`	No	`None`	Description of expected output shape.
`mcp_servers`	`list[MCPServerDetails]`	No	`[]`	Per-task MCP servers (in addition to the agent’s configured servers).
`triggering_agent_id`	`str`	No	`None`	ID of the agent that triggered this task (for sub-agent tracking).
`title`	`str`	No	`None`	Display title in the dashboard.
`think_mode`	`ThinkMode`	No	`Default`	`Default` or `Harder`. Toggles extended reasoning.
`disable_attachment_injection`	`bool`	No	`False`	Skip auto-injecting human-readable file content into the prompt.
`return_metrics`	`bool`	No	`False`	Return metrics in the task. Only valid for Workflow → Agent invocation.
`user_tokens`	`dict`	No	`None`	Pre-computed user tokens injected for MCP auth.

Returns `Task`

The created Task. See Task class reference. The task starts in status Pending. It moves through Executing → Completed (or Error / Failed / Stopped) as the worker runs it.

Examples

With files

task = await agent.acreate_task(
    prompt="Find action items in this meeting.",
    file_urls=[
        "https://example.com/recording.mp3",
        "https://example.com/transcript.pdf",
    ],
)

Structured output

from xpander_sdk import OutputFormat

task = await agent.acreate_task(
    prompt="Extract company name and revenue from the press release.",
    file_urls=["https://example.com/press-release.pdf"],
    output_format=OutputFormat.Json,
    output_schema={
        "type": "object",
        "properties": {
            "company": {"type": "string"},
            "revenue_usd": {"type": "number"},
        },
        "required": ["company"],
    },
)

Streaming

task = await agent.acreate_task(
    prompt="Long-running analysis…",
    events_streaming=True,
)

async for event in task.aevents():
    print(event.type, event.data)

events_streaming=True is required to use task.aevents(). See task.events.

Continue a thread

task = await agent.acreate_task(
    existing_task_id="task_abc123",
    prompt="What was the conclusion?",
)

The new task reuses the prior task’s memory thread, so the agent has full context.

Per-end-user

from xpander_sdk import User

task = await agent.acreate_task(
    prompt="Summarize my unread email.",
    user_details=User(
        id="user_42",
        email="alex@example.com",
        first_name="Alex",
        timezone="America/New_York",
    ),
)

user_details.id scopes user-memory and connector auth (e.g. each end user’s Gmail OAuth token).

Add an MCP server just for this task

from xpander_sdk import MCPServerDetails, MCPServerType, MCPServerTransport

task = await agent.acreate_task(
    prompt="Query our internal docs.",
    mcp_servers=[
        MCPServerDetails(
            type=MCPServerType.Remote,
            name="internal-docs",
            url="https://mcp.internal.acme.com",
            transport=MCPServerTransport.HTTP_Transport,
        ),
    ],
)

Per-task MCP servers stack on top of the agent’s persistent MCP configuration.

Pin a version

task = await agent.acreate_task(
    prompt="Run with the locked v3 prompt",
    agent_version="3",
)

Sync version

task = agent.create_task(prompt="Hello!")

Same parameters; blocks until the task is created. Don’t call from inside an event loop.

Errors

Raises ModuleException on failure:

Status	Cause
400	Invalid input (bad output schema, malformed `mcp_servers`, etc.).
401 / 403	Auth failure.
404	Agent doesn’t exist.
500	Server error.

​Parameters

​Returns Task

​Examples

​With files

​Structured output

​Streaming

​Continue a thread

​Per-end-user

​Add an MCP server just for this task

​Pin a version

​Sync version

​Errors