harnest.a2a

All Python modules · Import and compatibility guidance

Client operations

Method	Network behavior
`connect()`	Fetch and validate the Agent Card once
`send(...)`	Send one message; optionally poll only the returned Task
`stream(...)`	Stream one new interaction; requires `streaming=True` on the client
`get_task(task_id)`	Fetch one explicitly named Task
`list_tasks(...)`	Fetch one bounded, filtered page on request
`cancel_task(task_id)`	Request cancellation of one explicitly named Task
`subscribe(task_id)`	Subscribe to one existing Task

A2A event fields

The A2A contextId maps to the Harnest session ID. A text-input root accepts A2A text parts; a typed-input root requires exactly one structured data part. Harnest projects customer-visible text and structured results back to A2A parts while keeping tool arguments and results inside the executing agent boundary regardless of tool_activity. With OutputPolicy(thinking=True), a streaming Task emits provider-exposed reasoning as a working-status message whose metadata.harnest.type is thinking; reasoning is otherwise suppressed. Agent and graph-node lifecycle updates use the same working status with metadata.harnest.type: agent_activity; model metadata uses metadata.harnest.type: agent_metadata. Set agent_metadata=AgentMetadataMode.SUPPRESS to omit those updates and aggregate token counts from the final Message or Task artifact. Answer artifacts can carry metadata.harnest.agent attribution. With OutputPolicy(decision_results=True), streaming tasks also publish working-status metadata with metadata.harnest.type: decision_result and the typed decision in metadata.harnest.value. Decision results are hidden by default.

Exports

Name	Kind
`A2AClient`	Class
`A2AClientError`	Class
`A2AResult`	Class
`A2AUpdate`	Class
`RemoteAgent`	Class
`RemoteAgentError`	Class

A2AClient

Discover and invoke one remote agent without background polling or listing.

A2AClient(
    card_url: str,
    *,
    card: Mapping[str, Any] | None = None,
    timeout: float = 300,
    poll_interval: float = 0.5,
    streaming: bool = False,
    allowed_hosts: Sequence[str] = (),
    allow_insecure: bool = False,
    http_client: httpx.AsyncClient | None = None,
) -> None

Parameters and example

Parameter	Type	Default
`card_url`	`str`	Required
`card`	`Mapping[str, Any] \| None`	`None`
`timeout`	`float`	`300`
`poll_interval`	`float`	`0.5`
`streaming`	`bool`	`False`
`allowed_hosts`	`Sequence[str]`	`()`
`allow_insecure`	`bool`	`False`
`http_client`	`httpx.AsyncClient \| None`	`None`

from harnest.a2a import A2AClient

client = A2AClient(card_url=...)

A2AClient.card

Return the validated cached card without triggering network I/O.

Example

from harnest.a2a import A2AClient

client: A2AClient
value = client.card

A2AClient.connect

async A2AClient.connect() -> AgentCard

Resolve and validate the Agent Card exactly once for this client.

Example

from harnest.a2a import A2AClient

client: A2AClient
result = await client.connect()

A2AClient.send

async A2AClient.send(
    value: Any,
    *,
    context_id: str | None = None,
    task_id: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    accepted_output_modes: Sequence[str] = (),
    return_immediately: bool = False,
    wait: bool = True,
    credential: Credential | None = None,
) -> A2AResult

Send one message and poll only when the caller requested a final result.

Parameters and example

Parameter	Type	Default
`value`	`Any`	Required
`context_id`	`str \| None`	`None`
`task_id`	`str \| None`	`None`
`metadata`	`Mapping[str, Any] \| None`	`None`
`accepted_output_modes`	`Sequence[str]`	`()`
`return_immediately`	`bool`	`False`
`wait`	`bool`	`True`
`credential`	`Credential \| None`	`None`

from harnest.a2a import A2AClient

client: A2AClient
result = await client.send(value=...)

A2AClient.stream

A2AClient.stream(
    value: Any,
    *,
    context_id: str | None = None,
    task_id: str | None = None,
    metadata: Mapping[str, Any] | None = None,
    credential: Credential | None = None,
) -> AsyncIterator[A2AUpdate]

Yield remote updates without issuing any implicit follow-up operations.

Parameters and example

Parameter	Type	Default
`value`	`Any`	Required
`context_id`	`str \| None`	`None`
`task_id`	`str \| None`	`None`
`metadata`	`Mapping[str, Any] \| None`	`None`
`credential`	`Credential \| None`	`None`

from harnest.a2a import A2AClient

client: A2AClient
async for item in client.stream(value=...):
    ...

A2AClient.get_task

async A2AClient.get_task(task_id: str, *, history_length: int = 0, credential: Credential | None = None) -> Task

Fetch one task only after an explicit caller request.

Parameters and example

Parameter	Type	Default
`task_id`	`str`	Required
`history_length`	`int`	`0`
`credential`	`Credential \| None`	`None`

from harnest.a2a import A2AClient

client: A2AClient
result = await client.get_task(task_id=...)

A2AClient.list_tasks

async A2AClient.list_tasks(
    *,
    page_size: int = 50,
    page_token: str = '',
    context_id: str = '',
    include_artifacts: bool = False,
    credential: Credential | None = None,
) -> ListTasksResponse

List a bounded page only when directly requested by application code.

Parameters and example

Parameter	Type	Default
`page_size`	`int`	`50`
`page_token`	`str`	`''`
`context_id`	`str`	`''`
`include_artifacts`	`bool`	`False`
`credential`	`Credential \| None`	`None`

from harnest.a2a import A2AClient

client: A2AClient
result = await client.list_tasks()

A2AClient.cancel_task

async A2AClient.cancel_task(task_id: str, *, credential: Credential | None = None) -> Task

Cancel one task only in response to explicit local cancellation.

Parameters and example

Parameter	Type	Default
`task_id`	`str`	Required
`credential`	`Credential \| None`	`None`

from harnest.a2a import A2AClient

client: A2AClient
result = await client.cancel_task(task_id=...)

A2AClient.subscribe(task_id: str, *, credential: Credential | None = None) -> AsyncIterator[A2AUpdate]

Subscribe to an existing task without fetching or listing others.

Parameters and example

Parameter	Type	Default
`task_id`	`str`	Required
`credential`	`Credential \| None`	`None`

from harnest.a2a import A2AClient

client: A2AClient
async for item in client.subscribe(task_id=...):
    ...

A2AClient.close

async A2AClient.close() -> None

Close the selected transport or the unused discovery client once.

Example

from harnest.a2a import A2AClient

client: A2AClient
await client.close()

A2AClient.aenter

async A2AClient.__aenter__() -> 'A2AClient'

Example

from harnest.a2a import A2AClient

client: A2AClient
async with client:
    ...

A2AClient.aexit

async A2AClient.__aexit__() -> None

Example

from harnest.a2a import A2AClient

client: A2AClient
async with client:
    ...

A2AClientError

An outbound A2A interaction failed Harnest policy or protocol handling.

Example

from harnest.a2a import A2AClientError

try:
    ...
except A2AClientError as error:
    ...

A2AResult

Minimum portable projection of a direct Message or tracked A2A Task.

A2AResult(
    text: str,
    data: Any,
    context_id: str | None,
    task_id: str | None,
    state: str,
    message: Message | None = None,
    task: Task | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`text`	`str`	`—`
`data`	`Any`	`—`
`context_id`	`str \| None`	`—`
`task_id`	`str \| None`	`—`
`state`	`str`	`—`
`message`	`Message \| None`	`None`
`task`	`Task \| None`	`None`

Parameters and example

Parameter	Type	Default
`text`	`str`	Required
`data`	`Any`	Required
`context_id`	`str \| None`	Required
`task_id`	`str \| None`	Required
`state`	`str`	Required
`message`	`Message \| None`	`None`
`task`	`Task \| None`	`None`

from harnest.a2a import A2AResult

result = A2AResult(text=..., data=..., context_id=..., task_id=..., state=...)

A2AResult.terminal

Return whether the remote interaction no longer needs observation.

Example

from harnest.a2a import A2AResult

result: A2AResult
value = result.terminal

A2AResult.as_dict

A2AResult.as_dict() -> dict[str, Any]

Serialize only portable output and opaque task references.

Example

from harnest.a2a import A2AResult

result: A2AResult
result = result.as_dict()

A2AUpdate

One normalized outbound stream update without protobuf coupling.

A2AUpdate(
    kind: str,
    task_id: str | None,
    context_id: str | None,
    state: str | None = None,
    text: str = '',
    data: Any = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`kind`	`str`	`—`
`task_id`	`str \| None`	`—`
`context_id`	`str \| None`	`—`
`state`	`str \| None`	`None`
`text`	`str`	`''`
`data`	`Any`	`None`

Parameters and example

Parameter	Type	Default
`kind`	`str`	Required
`task_id`	`str \| None`	Required
`context_id`	`str \| None`	Required
`state`	`str \| None`	`None`
`text`	`str`	`''`
`data`	`Any`	`None`

from harnest.a2a import A2AUpdate

update = A2AUpdate(kind=..., task_id=..., context_id=...)

RemoteAgent

A first-class remote A2A participant usable as a portable graph node.

RemoteAgent(
    name: str,
    card_url: str,
    description: str = '',
    audience: str | None = None,
    scopes: tuple[str, ...] = (),
    timeout: float = 300,
    allowed_hosts: tuple[str, ...] = (),
    allow_insecure: bool = False,
    card: Mapping[str, Any] | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`name`	`str`	`—`
`card_url`	`str`	`—`
`description`	`str`	`''`
`audience`	`str \| None`	`None`
`scopes`	`tuple[str, ...]`	`()`
`timeout`	`float`	`300`
`allowed_hosts`	`tuple[str, ...]`	`()`
`allow_insecure`	`bool`	`False`
`card`	`Mapping[str, Any] \| None`	`None`

Parameters and example

Parameter	Type	Default
`name`	`str`	Required
`card_url`	`str`	Required
`description`	`str`	`''`
`audience`	`str \| None`	`None`
`scopes`	`tuple[str, ...]`	`()`
`timeout`	`float`	`300`
`allowed_hosts`	`tuple[str, ...]`	`()`
`allow_insecure`	`bool`	`False`
`card`	`Mapping[str, Any] \| None`	`None`

from harnest.a2a import RemoteAgent

agent = RemoteAgent(name=..., card_url=...)

RemoteAgent.call

async RemoteAgent.__call__(value: Any, *, context: GraphContext) -> Event

Delegate one graph value and retain only opaque conversation references.

Parameters and example

Parameter	Type	Default
`value`	`Any`	Required
`context`	`GraphContext`	Required

from harnest.a2a import RemoteAgent

agent: RemoteAgent
event = await agent(value=..., context=...)

RemoteAgent.as_tool

RemoteAgent.as_tool() -> Any

Expose delegation as an explicit text tool when agent semantics are unsuitable.

Example

from harnest.a2a import RemoteAgent

agent: RemoteAgent
result = agent.as_tool()

RemoteAgentError

A remote agent reached a terminal failure state.

Example

from harnest.a2a import RemoteAgentError

try:
    ...
except RemoteAgentError as error:
    ...

Overview

Get started

Harnest Studio

Guides

API reference

Client operations

A2A event fields

Exports

A2AClient

A2AClient.card

A2AClient.connect

A2AClient.send

A2AClient.stream

A2AClient.get_task

A2AClient.list_tasks

A2AClient.cancel_task

A2AClient.close

A2AClient.aenter

A2AClient.aexit

A2AClientError

A2AResult

A2AResult.terminal

A2AResult.as_dict

A2AUpdate

RemoteAgent

RemoteAgent.call

RemoteAgent.as_tool

RemoteAgentError

​Client operations

​A2A event fields

​Exports

​A2AClient

​A2AClient.card

​A2AClient.connect

​A2AClient.send

​A2AClient.stream

​A2AClient.get_task

​A2AClient.list_tasks

​A2AClient.cancel_task

​A2AClient.subscribe

​A2AClient.close

​A2AClient.aenter

​A2AClient.aexit

​A2AClientError

​A2AResult

​A2AResult.terminal

​A2AResult.as_dict

​A2AUpdate

​RemoteAgent

​RemoteAgent.call

​RemoteAgent.as_tool

​RemoteAgentError

Client operations

A2A event fields

Exports

A2AClient

A2AClient.card

A2AClient.connect

A2AClient.send

A2AClient.stream

A2AClient.get_task

A2AClient.list_tasks

A2AClient.cancel_task

A2AClient.subscribe

A2AClient.close

A2AClient.aenter

A2AClient.aexit

A2AClientError

A2AResult

A2AResult.terminal

A2AResult.as_dict

A2AUpdate

RemoteAgent

RemoteAgent.call

RemoteAgent.as_tool

RemoteAgentError