TaskContext API Reference¶

TaskContext is the execution context passed to Worker.handle(). It provides the complete interface for interacting with tasks — reading input, emitting results, controlling lifecycle, and accessing dependencies.

`a2akit.worker.base.TaskContext` ¶

Bases: ABC

Execution context passed to Worker.handle().

Attributes:

Name	Type	Description
`task_id`	`str`	Current task identifier.
`context_id`	`str \| None`	Optional conversation / context identifier.
`message_id`	`str`	Identifier of the triggering message.
`user_text`	`str`	The user's input as plain text.
`parts`	`list[Any]`	Raw message parts (text, files, etc.).
`metadata`	`dict[str, Any]`	Arbitrary metadata forwarded from the request.

`user_text` `instance-attribute` ¶

`parts` `instance-attribute` ¶

`files` `abstractmethod` `property` ¶

All file parts from the user message as typed wrappers.

`data_parts` `abstractmethod` `property` ¶

All structured data parts from the user message.

`task_id` `instance-attribute` ¶

`context_id` `instance-attribute` ¶

`message_id` `instance-attribute` ¶

`metadata` `instance-attribute` ¶

`request_context` `abstractmethod` `property` ¶

Transient request data from middleware.

Contains secrets, headers, and other per-request data that was NOT persisted to Storage. Populated by middleware in before_dispatch. Empty dict if no middleware is registered.

This is separate from self.metadata which contains persisted data from message.metadata.

`is_cancelled` `abstractmethod` `property` ¶

Check whether cancellation has been requested for this task.

`turn_ended` `abstractmethod` `property` ¶

Whether the handler signaled the end of this processing turn.

True after calling a terminal method (complete/fail/reject/respond) or requesting further input (request_input/request_auth).

`history` `abstractmethod` `property` ¶

Previous messages within this task (excluding the current message).

`previous_artifacts` `abstractmethod` `property` ¶

Artifacts already produced by this task in previous turns.

`deps` `abstractmethod` `property` ¶

Dependency container registered on the server.

Access by type key or string key::

db = ctx.deps[DatabasePool]
key = ctx.deps.get("api_key", "default")

`complete(text=None, *, artifact_id='final-answer')` `abstractmethod` `async` ¶

Mark the task as completed, optionally with a final text artifact.

`complete_json(data, *, artifact_id='final-answer')` `abstractmethod` `async` ¶

Complete task with a JSON data artifact.

`fail(reason)` `abstractmethod` `async` ¶

Mark the task as failed with an error reason.

`reject(reason=None)` `abstractmethod` `async` ¶

Reject the task — agent decides not to perform it.

`request_input(question)` `abstractmethod` `async` ¶

Transition to input-required state.

`request_auth(details=None)` `abstractmethod` `async` ¶

Transition to auth-required state for secondary credentials.

`respond(text=None)` `abstractmethod` `async` ¶

Complete the task with a direct message response (no artifact created).

`reply_directly(text)` `abstractmethod` `async` ¶

Return a Message directly without task tracking.

The HTTP response to message:send will be {"message": {...}} instead of {"task": {...}}. The task is still created internally for lifecycle management but the client receives only the message.

Streaming note: When the request arrives via message:stream, this method uses the Task lifecycle stream pattern (Task snapshot followed by events and a terminal status update) rather than a message-only stream. This is spec-conformant (§3.1.2) — servers may always respond with a Task lifecycle stream.

`send_status(message=None)` `abstractmethod` `async` ¶

Emit an intermediate status update (state stays working).

When message is provided, the status message is persisted in Storage so that polling clients (tasks/get) can see progress — not just SSE subscribers. The message is stored in task.status.message only (NOT appended to history) to keep the write lightweight.

When message is None, only a bare working-state event is broadcast (no storage write).

`emit_artifact(*, artifact_id, text=None, data=None, file_bytes=None, file_url=None, media_type=None, filename=None, name=None, description=None, append=False, last_chunk=False, metadata=None)` `abstractmethod` `async` ¶

Emit an artifact update event and persist it.

`emit_text_artifact(text, *, artifact_id='answer', append=False, last_chunk=False)` `abstractmethod` `async` ¶

Emit a single-text artifact chunk.

`emit_data_artifact(data, *, artifact_id='answer', media_type='application/json', append=False, last_chunk=False)` `abstractmethod` `async` ¶

Emit a structured data artifact chunk.

`load_context()` `abstractmethod` `async` ¶

Load stored context for this task's context_id.

`update_context(context)` `abstractmethod` `async` ¶

Store context for this task's context_id.

Properties¶

Input¶

Property	Type	Description
`user_text`	`str`	The user's input as plain text (all text parts joined)
`parts`	`list[Any]`	Raw A2A message parts (text, files, data)
`files`	`list[FileInfo]`	File parts as typed wrappers
`data_parts`	`list[dict]`	Structured data parts extracted from the message

Identifiers¶

Property	Type	Description
`task_id`	`str`	Current task UUID
`context_id`	`str \\| None`	Conversation / context identifier
`message_id`	`str`	ID of the triggering message

Metadata¶

Property	Type	Description
`metadata`	`dict[str, Any]`	Persisted metadata from `message.metadata`
`request_context`	`dict[str, Any]`	Transient data from middleware (never persisted)

State¶

Property	Type	Description
`is_cancelled`	`bool`	Whether cancellation was requested
`turn_ended`	`bool`	Whether a lifecycle method was called

History¶

Property	Type	Description
`history`	`list[HistoryMessage]`	Previous messages in this task
`previous_artifacts`	`list[PreviousArtifact]`	Artifacts from prior turns

Dependencies¶

Property	Type	Description
`deps`	`DependencyContainer`	Dependency container from the server

Lifecycle Methods¶

These methods transition the task to a new state. Exactly one must be called per handle() invocation.

`complete(text=None, *, artifact_id="final-answer")`¶

Mark the task as completed. Optionally attach a text artifact.

await ctx.complete("Here is your answer.")
await ctx.complete()  # completed without artifact

`complete_json(data, *, artifact_id="final-answer")`¶

Complete with a JSON data artifact.

await ctx.complete_json({"score": 0.95, "label": "positive"})

`respond(text=None)`¶

Complete with a status message only — no artifact is created.

await ctx.respond("Task acknowledged.")

`reply_directly(text)`¶

Return a Message directly without task tracking. The HTTP response to message:send will be {"message": {...}} instead of {"task": {...}}.

await ctx.reply_directly("Quick answer without task overhead.")

`fail(reason)`¶

Mark the task as failed.

await ctx.fail("External API returned 500.")

`reject(reason=None)`¶

Reject the task — the agent decides not to perform it.

await ctx.reject("I can only process English text.")

`request_input(question)`¶

Transition to input-required. The client should send a follow-up message.

await ctx.request_input("Which language should I translate to?")

`request_auth(details=None)`¶

Transition to auth-required for secondary credentials.

await ctx.request_auth("Please provide your GitHub token.")

Streaming Methods¶

These methods emit events while the task stays in working state.

`send_status(message=None)`¶

Emit an intermediate status update. When message is provided, it's persisted in task.status.message.

await ctx.send_status("Processing 3 of 10 files...")

`emit_text_artifact(text, *, artifact_id="answer", append=False, last_chunk=False)`¶

Emit a text chunk as an artifact update.

`emit_data_artifact(data, *, artifact_id="answer", media_type="application/json", append=False, last_chunk=False)`¶

Emit structured data as an artifact update.

`emit_artifact(*, artifact_id, text=None, data=None, file_bytes=None, file_url=None, media_type=None, filename=None, name=None, description=None, append=False, last_chunk=False, metadata=None)`¶

General-purpose artifact emission supporting text, data, file bytes, and file URLs.

Context Methods¶

`load_context()`¶

Load stored context for this task's context_id. Returns None if no context exists or context_id is None.

`update_context(context)`¶

Store context for this task's context_id. No-op if context_id is None.

Helper Types¶

FileInfo¶

@dataclass(frozen=True)
class FileInfo:
    content: bytes | None
    url: str | None
    filename: str | None
    media_type: str | None

HistoryMessage¶

@dataclass(frozen=True)
class HistoryMessage:
    role: str
    text: str
    parts: list[Any]
    message_id: str

PreviousArtifact¶

@dataclass(frozen=True)
class PreviousArtifact:
    artifact_id: str
    name: str | None
    parts: list[Any]

TaskContext API Reference¶

a2akit.worker.base.TaskContext ¶

user_text instance-attribute ¶

parts instance-attribute ¶

files abstractmethod property ¶

data_parts abstractmethod property ¶

task_id instance-attribute ¶

context_id instance-attribute ¶

message_id instance-attribute ¶

metadata instance-attribute ¶

request_context abstractmethod property ¶

is_cancelled abstractmethod property ¶

turn_ended abstractmethod property ¶

history abstractmethod property ¶

previous_artifacts abstractmethod property ¶

deps abstractmethod property ¶

complete(text=None, *, artifact_id='final-answer') abstractmethod async ¶

complete_json(data, *, artifact_id='final-answer') abstractmethod async ¶

fail(reason) abstractmethod async ¶

reject(reason=None) abstractmethod async ¶

request_input(question) abstractmethod async ¶

request_auth(details=None) abstractmethod async ¶

respond(text=None) abstractmethod async ¶

reply_directly(text) abstractmethod async ¶

send_status(message=None) abstractmethod async ¶

emit_artifact(*, artifact_id, text=None, data=None, file_bytes=None, file_url=None, media_type=None, filename=None, name=None, description=None, append=False, last_chunk=False, metadata=None) abstractmethod async ¶

emit_text_artifact(text, *, artifact_id='answer', append=False, last_chunk=False) abstractmethod async ¶

emit_data_artifact(data, *, artifact_id='answer', media_type='application/json', append=False, last_chunk=False) abstractmethod async ¶

load_context() abstractmethod async ¶

update_context(context) abstractmethod async ¶