Remote execution

Types for invoking remote nodes and tracking remote task lifecycles, including streaming events, pagination, and terminal-state constants.

remote

Remote transport protocol and helper node for PenguiFlow.

REMOTE_TERMINAL_TASK_STATES module-attribute

REMOTE_TERMINAL_TASK_STATES = frozenset({RemoteTaskState.COMPLETED, RemoteTaskState.FAILED, RemoteTaskState.CANCELLED, RemoteTaskState.REJECTED})

RemoteTransport

Bases: Protocol

Protocol describing the minimal remote invocation surface.

Implement this protocol to plug a custom remote-agent transport (e.g. HTTP, gRPC, A2A) into :func:RemoteNode. Only send, stream, and cancel are required; richer task-oriented operations belong to :class:SupportsRemoteTasks.

send async

send(request: RemoteCallRequest) -> RemoteCallResult

Perform a unary remote call.

Parameters:

Name	Type	Description	Default
request	RemoteCallRequest	The call parameters, including the message, skill, and target agent URL.	required

Returns:

Type	Description
RemoteCallResult	The remote agent's result wrapped in a :class:RemoteCallResult.

Raises:

Type	Description
Exception	Implementations should raise on transport failure; PenguiFlow logs the error via remote_call_error events and re-raises it to the caller.

stream

stream(request: RemoteCallRequest) -> AsyncIterator[RemoteStreamEvent]

Perform a remote call that yields streaming events.

Parameters:

Name	Type	Description	Default
request	RemoteCallRequest	The call parameters, including the message, skill, and target agent URL.	required

Yields:

Type	Description
AsyncIterator[RemoteStreamEvent]	class:RemoteStreamEvent instances as the remote agent produces output; the final
AsyncIterator[RemoteStreamEvent]	event should set done=True and may include a terminal result.

Raises:

Type	Description
Exception	Implementations should raise on transport failure.

cancel async

cancel(*, agent_url: str, task_id: str) -> None

Cancel a remote task identified by task_id at agent_url.

Parameters:

Name	Type	Description	Default
agent_url	str	Base URL of the remote agent that owns the task.	required
task_id	str	Identifier of the remote task to cancel.	required

Raises:

Type	Description
Exception	Implementations should raise on transport failure; callers in this module log but otherwise swallow cancellation errors defensively.

RemoteCallRequest dataclass

RemoteCallRequest(message: Message, skill: str, agent_url: str, agent_card: Mapping[str, Any] | None = None, metadata: Mapping[str, Any] | None = None, timeout_s: float | None = None, context_id: str | None = None, task_id: str | None = None)

Input to :class:RemoteTransport implementations.

Attributes:

Name	Type	Description
message	Message	The PenguiFlow :class:~penguiflow.types.Message being forwarded to the remote agent.
skill	str	Name of the remote skill/capability to invoke.
agent_url	str	Base URL of the remote agent to call.
agent_card	Mapping[str, Any] | None	Optional agent card metadata (e.g. capabilities descriptor) for the target agent.
metadata	Mapping[str, Any] | None	Optional request metadata forwarded alongside the message (typically message.meta).
timeout_s	float | None	Optional per-call timeout in seconds; None means no explicit timeout override.
context_id	str | None	Optional remote conversation/context identifier to continue an existing context.
task_id	str | None	Optional remote task identifier to continue or resume an existing task.

message instance-attribute

message: Message

skill instance-attribute

skill: str

agent_url instance-attribute

agent_url: str

agent_card class-attribute instance-attribute

agent_card: Mapping[str, Any] | None = None

metadata class-attribute instance-attribute

metadata: Mapping[str, Any] | None = None

timeout_s class-attribute instance-attribute

timeout_s: float | None = None

context_id class-attribute instance-attribute

context_id: str | None = None

task_id class-attribute instance-attribute

task_id: str | None = None

RemoteCallResult dataclass

RemoteCallResult(result: Any, context_id: str | None = None, task_id: str | None = None, agent_url: str | None = None, meta: Mapping[str, Any] | None = None)

Return value for :meth:RemoteTransport.send.

Attributes:

Name	Type	Description
result	Any	The unary result payload returned by the remote agent.
context_id	str | None	Remote conversation/context identifier assigned or reused by the remote agent, if any.
task_id	str | None	Remote task identifier assigned by the remote agent, if any.
agent_url	str | None	Agent URL that actually served the request (may differ from the requested agent_url).
meta	Mapping[str, Any] | None	Optional additional metadata returned by the transport.

result instance-attribute

result: Any

context_id class-attribute instance-attribute

context_id: str | None = None

task_id class-attribute instance-attribute

task_id: str | None = None

agent_url class-attribute instance-attribute

agent_url: str | None = None

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = None

RemotePushNotificationBinding dataclass

RemotePushNotificationBinding(name: str | None, config: Mapping[str, Any])

Normalized push notification config returned by remote transports.

Attributes:

Name	Type	Description
name	str | None	Optional identifier/name of the push notification configuration.
config	Mapping[str, Any]	The push notification configuration payload (transport-specific shape).

name instance-attribute

name: str | None

config instance-attribute

config: Mapping[str, Any]

RemoteStreamEvent dataclass

RemoteStreamEvent(text: str | None = None, done: bool = False, meta: Mapping[str, Any] | None = None, context_id: str | None = None, task_id: str | None = None, agent_url: str | None = None, result: Any | None = None)

Streaming event yielded by :meth:RemoteTransport.stream.

Attributes:

Name	Type	Description
text	str | None	Incremental text chunk emitted by the remote agent, if any.
done	bool	Whether this event marks the end of the stream.
meta	Mapping[str, Any] | None	Optional metadata associated with this chunk (e.g. token usage, reasoning tags).
context_id	str | None	Remote conversation/context identifier associated with this event, if known.
task_id	str | None	Remote task identifier associated with this event, if known.
agent_url	str | None	Agent URL that produced this event, if known (may change across retries/redirects).
result	Any | None	Optional final result payload attached to a terminal event.

text class-attribute instance-attribute

text: str | None = None

done class-attribute instance-attribute

done: bool = False

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = None

context_id class-attribute instance-attribute

context_id: str | None = None

task_id class-attribute instance-attribute

task_id: str | None = None

agent_url class-attribute instance-attribute

agent_url: str | None = None

result class-attribute instance-attribute

result: Any | None = None

RemoteTaskEvent dataclass

RemoteTaskEvent(kind: str, task: RemoteTaskSnapshot | None = None, status: RemoteTaskStatus | None = None, text: str | None = None, result: Any | None = None, artifact: Any | None = None, done: bool = False, context_id: str | None = None, task_id: str | None = None, agent_url: str | None = None, meta: Mapping[str, Any] | None = None, raw: Mapping[str, Any] | None = None)

Normalized event emitted by task subscriptions.

Attributes:

Name	Type	Description
kind	str	Discriminator describing the kind of event (transport-specific string).
task	RemoteTaskSnapshot | None	Optional full task snapshot attached to this event.
status	RemoteTaskStatus | None	Optional status update attached to this event.
text	str | None	Optional incremental text chunk attached to this event.
result	Any | None	Optional result payload attached to this event.
artifact	Any | None	Optional artifact attached to this event.
done	bool	Whether this event marks the end of the subscription stream.
context_id	str | None	Remote conversation/context identifier associated with this event, if known.
task_id	str | None	Remote task identifier associated with this event, if known.
agent_url	str | None	Agent URL that produced this event, if known.
meta	Mapping[str, Any] | None	Optional additional metadata for this event.
raw	Mapping[str, Any] | None	Optional raw, transport-specific event payload for debugging.

kind instance-attribute

kind: str

task class-attribute instance-attribute

task: RemoteTaskSnapshot | None = None

status class-attribute instance-attribute

status: RemoteTaskStatus | None = None

text class-attribute instance-attribute

text: str | None = None

result class-attribute instance-attribute

result: Any | None = None

artifact class-attribute instance-attribute

artifact: Any | None = None

done class-attribute instance-attribute

done: bool = False

context_id class-attribute instance-attribute

context_id: str | None = None

task_id class-attribute instance-attribute

task_id: str | None = None

agent_url class-attribute instance-attribute

agent_url: str | None = None

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = None

raw class-attribute instance-attribute

raw: Mapping[str, Any] | None = None

RemoteTaskAuthRequired

RemoteTaskAuthRequired(snapshot: RemoteTaskSnapshot)

Bases: RuntimeError

Raised when a remote task requires authorization before it can continue.

Attributes:

Name	Type	Description
snapshot		The :class:RemoteTaskSnapshot that triggered this exception, carrying the remote task's current state (including any status message).

Initialize the exception from a remote task snapshot.

Parameters:

Name	Type	Description	Default
snapshot	RemoteTaskSnapshot	Task snapshot whose status indicated AUTH_REQUIRED.	required

snapshot instance-attribute

snapshot = snapshot

RemoteTaskInputRequired

RemoteTaskInputRequired(snapshot: RemoteTaskSnapshot)

Bases: RuntimeError

Raised when a remote task needs user clarification before it can continue.

Attributes:

Name	Type	Description
snapshot		The :class:RemoteTaskSnapshot that triggered this exception, carrying the remote task's current state (including any status message).

Initialize the exception from a remote task snapshot.

Parameters:

Name	Type	Description	Default
snapshot	RemoteTaskSnapshot	Task snapshot whose status indicated INPUT_REQUIRED.	required

snapshot instance-attribute

snapshot = snapshot

RemoteTaskPage dataclass

RemoteTaskPage(tasks: list[RemoteTaskSnapshot], next_page_token: str = '', page_size: int = 0, total_size: int = 0)

Paginated task list response.

Attributes:

Name	Type	Description
tasks	list[RemoteTaskSnapshot]	Page of normalized task snapshots.
next_page_token	str	Opaque token to fetch the next page; empty string if there is no next page.
page_size	int	Number of items requested per page.
total_size	int	Total number of tasks available across all pages, if reported by the transport.

tasks instance-attribute

tasks: list[RemoteTaskSnapshot]

next_page_token class-attribute instance-attribute

next_page_token: str = ''

page_size class-attribute instance-attribute

page_size: int = 0

total_size class-attribute instance-attribute

total_size: int = 0

RemoteTaskSnapshot dataclass

RemoteTaskSnapshot(task_id: str, context_id: str, status: RemoteTaskStatus, result: Any | None = None, artifacts: list[Any] | None = None, history: list[Any] | None = None, agent_url: str | None = None, meta: Mapping[str, Any] | None = None)

Normalized task representation returned by task-oriented remote transports.

Attributes:

Name	Type	Description
task_id	str	Remote task identifier.
context_id	str	Remote conversation/context identifier the task belongs to.
status	RemoteTaskStatus	Current normalized status of the task.
result	Any | None	Optional result payload once the task has produced output.
artifacts	list[Any] | None	Optional list of artifacts produced by the task.
history	list[Any] | None	Optional list of historical events/messages for the task.
agent_url	str | None	Optional agent URL that owns this task.
meta	Mapping[str, Any] | None	Optional additional metadata about the task.

task_id instance-attribute

task_id: str

context_id instance-attribute

context_id: str

status instance-attribute

status: RemoteTaskStatus

result class-attribute instance-attribute

result: Any | None = None

artifacts class-attribute instance-attribute

artifacts: list[Any] | None = None

history class-attribute instance-attribute

history: list[Any] | None = None

agent_url class-attribute instance-attribute

agent_url: str | None = None

meta class-attribute instance-attribute

meta: Mapping[str, Any] | None = None

is_terminal property

is_terminal: bool

Return whether status.state is one of the terminal task states.

RemoteTaskState

Bases: str, Enum

Task lifecycle states exposed by task-oriented remote transports.

Attributes:

Name	Type	Description
UNSPECIFIED		State was not reported by the remote transport.
SUBMITTED		Task was accepted by the remote agent but has not started running yet.
WORKING		Task is actively being processed by the remote agent.
COMPLETED		Task finished successfully (terminal state).
FAILED		Task finished with an error (terminal state).
CANCELLED		Task was cancelled before completion (terminal state).
INPUT_REQUIRED		Task is paused pending additional user input.
AUTH_REQUIRED		Task is paused pending authorization/credentials.
REJECTED		Task was rejected by the remote agent (terminal state).

UNSPECIFIED class-attribute instance-attribute

UNSPECIFIED = 'unspecified'

SUBMITTED class-attribute instance-attribute

SUBMITTED = 'submitted'

WORKING class-attribute instance-attribute

WORKING = 'working'

COMPLETED class-attribute instance-attribute

COMPLETED = 'completed'

FAILED class-attribute instance-attribute

FAILED = 'failed'

CANCELLED class-attribute instance-attribute

CANCELLED = 'cancelled'

INPUT_REQUIRED class-attribute instance-attribute

INPUT_REQUIRED = 'input-required'

AUTH_REQUIRED class-attribute instance-attribute

AUTH_REQUIRED = 'auth-required'

REJECTED class-attribute instance-attribute

REJECTED = 'rejected'

RemoteTaskStatus dataclass

RemoteTaskStatus(state: RemoteTaskState, message: str | None = None, timestamp: str | None = None, raw: Mapping[str, Any] | None = None)

Normalized status for a remote task.

Attributes:

Name	Type	Description
state	RemoteTaskState	Current lifecycle state of the remote task.
message	str | None	Optional human-readable status message from the remote agent.
timestamp	str | None	Optional ISO-8601 timestamp of when this status was reported.
raw	Mapping[str, Any] | None	Optional raw, transport-specific status payload for debugging.

state instance-attribute

state: RemoteTaskState

message class-attribute instance-attribute

message: str | None = None

timestamp class-attribute instance-attribute

timestamp: str | None = None

raw class-attribute instance-attribute

raw: Mapping[str, Any] | None = None

SupportsRemoteTasks

Bases: Protocol

Optional task-oriented remote execution surface.

Implement this protocol in addition to :class:RemoteTransport when the remote agent supports long-running, resumable tasks (create/fetch/list/subscribe, plus push notification configuration). Runtime checks use isinstance(transport, SupportsRemoteTasks) to detect support for these operations.

send_task async

send_task(request: RemoteCallRequest, *, blocking: bool = False) -> RemoteTaskSnapshot

Create or continue a remote task and return its current snapshot.

Parameters:

Name	Type	Description	Default
request	RemoteCallRequest	The call parameters, including the message, skill, and target agent URL.	required
blocking	bool	If True, wait for the task to reach a terminal state before returning.	False

Returns:

Type	Description
RemoteTaskSnapshot	The current :class:RemoteTaskSnapshot for the created/continued task.

Raises:

Type	Description
Exception	Implementations should raise on transport failure.

get_task async

get_task(*, agent_url: str, task_id: str, history_length: int | None = None) -> RemoteTaskSnapshot

Fetch a remote task snapshot.

Parameters:

Name	Type	Description	Default
agent_url	str	Base URL of the remote agent that owns the task.	required
task_id	str	Identifier of the remote task to fetch.	required
history_length	int | None	Optional cap on the number of history entries to return.	None

Returns:

Type	Description
RemoteTaskSnapshot	The current :class:RemoteTaskSnapshot for the task.

Raises:

Type	Description
Exception	Implementations should raise if the task cannot be found or fetched.

list_tasks async

list_tasks(*, agent_url: str, context_id: str | None = None, status: RemoteTaskState | str | None = None, page_size: int = 100, page_token: str | None = None, history_length: int | None = None, include_artifacts: bool = False) -> RemoteTaskPage

List remote tasks, optionally filtered by context and status.

Parameters:

Name	Type	Description	Default
agent_url	str	Base URL of the remote agent to query.	required
context_id	str | None	Optional remote context identifier to filter by.	None
status	RemoteTaskState | str | None	Optional task state (or raw string) to filter by.	None
page_size	int	Maximum number of tasks to return per page.	100
page_token	str | None	Optional opaque token to resume from a previous page.	None
history_length	int | None	Optional cap on the number of history entries per task.	None
include_artifacts	bool	Whether to include task artifacts in the response.	False

Returns:

Name	Type	Description
A	RemoteTaskPage	class:RemoteTaskPage with the matching tasks and pagination info.

Raises:

Type	Description
Exception	Implementations should raise on transport failure.

subscribe_task

subscribe_task(*, agent_url: str, task_id: str) -> AsyncIterator[RemoteTaskEvent]

Subscribe to task updates.

Parameters:

Name	Type	Description	Default
agent_url	str	Base URL of the remote agent that owns the task.	required
task_id	str	Identifier of the remote task to subscribe to.	required

Yields:

Type	Description
AsyncIterator[RemoteTaskEvent]	class:RemoteTaskEvent instances as the remote task progresses; the final event
AsyncIterator[RemoteTaskEvent]	should set done=True.

Raises:

Type	Description
Exception	Implementations should raise on transport failure.

set_task_push_notification_config async

set_task_push_notification_config(*, agent_url: str, task_id: str, config_id: str, config: Mapping[str, Any]) -> RemotePushNotificationBinding

Set a push notification config for a remote task.

Parameters:

Name	Type	Description	Default
agent_url	str	Base URL of the remote agent that owns the task.	required
task_id	str	Identifier of the remote task to configure.	required
config_id	str	Identifier for the push notification configuration.	required
config	Mapping[str, Any]	The push notification configuration payload (transport-specific shape).	required

Returns:

Type	Description
RemotePushNotificationBinding	The stored :class:RemotePushNotificationBinding.

Raises:

Type	Description
Exception	Implementations should raise on transport failure.

get_task_push_notification_config async

get_task_push_notification_config(*, agent_url: str, task_id: str, config_id: str) -> RemotePushNotificationBinding

Fetch a push notification config for a remote task.

Parameters:

Name	Type	Description	Default
agent_url	str	Base URL of the remote agent that owns the task.	required
task_id	str	Identifier of the remote task.	required
config_id	str	Identifier of the push notification configuration to fetch.	required

Returns:

Type	Description
RemotePushNotificationBinding	The matching :class:RemotePushNotificationBinding.

Raises:

Type	Description
Exception	Implementations should raise if the configuration cannot be found.

list_task_push_notification_configs async

list_task_push_notification_configs(*, agent_url: str, task_id: str, page_size: int = 100, page_token: str | None = None) -> list[RemotePushNotificationBinding]

List push notification configs for a remote task.

Parameters:

Name	Type	Description	Default
agent_url	str	Base URL of the remote agent that owns the task.	required
task_id	str	Identifier of the remote task.	required
page_size	int	Maximum number of configurations to return per page.	100
page_token	str | None	Optional opaque token to resume from a previous page.	None

Returns:

Type	Description
list[RemotePushNotificationBinding]	A list of :class:RemotePushNotificationBinding entries.

Raises:

Type	Description
Exception	Implementations should raise on transport failure.

delete_task_push_notification_config async

delete_task_push_notification_config(*, agent_url: str, task_id: str, config_id: str) -> None

Delete a push notification config for a remote task.

Parameters:

Name	Type	Description	Default
agent_url	str	Base URL of the remote agent that owns the task.	required
task_id	str	Identifier of the remote task.	required
config_id	str	Identifier of the push notification configuration to delete.	required

Raises:

Type	Description
Exception	Implementations should raise on transport failure.

RemoteNode

RemoteNode(*, transport: RemoteTransport, skill: str, agent_url: str, name: str, agent_card: Mapping[str, Any] | None = None, policy: NodePolicy | None = None, streaming: bool = False, record_binding: bool = True) -> Node

Create a node that proxies work to a remote agent via transport.

The returned :class:~penguiflow.node.Node forwards incoming :class:~penguiflow.types.Message payloads to the remote agent, mirrors trace cancellation onto the remote task (when record_binding is enabled and the transport reports a task_id), and emits remote_call_*/remote_stream_event observability events via the owning runtime.

Parameters:

Name	Type	Description	Default
transport	RemoteTransport	The :class:RemoteTransport implementation used to call the remote agent.	required
skill	str	Name of the remote skill/capability to invoke.	required
agent_url	str	Base URL of the remote agent to call.	required
name	str	Name assigned to the returned node.	required
agent_card	Mapping[str, Any] | None	Optional agent card metadata forwarded with each request.	None
policy	NodePolicy | None	Optional :class:~penguiflow.node.NodePolicy; defaults to NodePolicy().	None
streaming	bool	If True, use transport.stream and forward chunks via ctx.emit_chunk; otherwise use transport.send for a single unary call.	False
record_binding	bool	If True, persist a remote binding for the trace (via runtime.save_remote_binding) and mirror trace-level cancellation onto the remote task.	True

Returns:

Name	Type	Description
A	Node	class:~penguiflow.node.Node wrapping the remote call.

Raises:

Type	Description
TypeError	If the node receives an input that is not a :class:~penguiflow.types.Message.
RuntimeError	If the node's context is not bound to a running :class:~penguiflow.core.PenguiFlow runtime, or the context owner is not a :class:~penguiflow.node.Node.