temporalio.contrib.deepagents._activity.DeepAgentActivities

class documentation

class DeepAgentActivities: (source)

Constructor: DeepAgentActivities(model_provider, streaming_batch_interval)

View In Hierarchy

Holds the worker-side dependencies and exposes the four activities.

An instance is created by DeepAgentsPlugin and its bound methods are registered on the worker.

Method __init__ Store the worker-side model provider + streaming configuration.
Async Method backend_op Run one operation against a registered (real-I/O) backend.
Async Method invoke_model Run exactly one LLM call and return the resulting AIMessage.
Async Method invoke_model_streaming Stream one LLM call, publishing chunk batches to streaming_topic.
Async Method invoke_tool Execute one registered tool and return its ToolMessage.
Method _build_bound_model Undocumented
Instance Variable _model_provider Undocumented
Instance Variable _streaming_batch_interval Undocumented
def __init__(self, *, model_provider: Callable[[str], Any] | None = None, streaming_batch_interval: timedelta = timedelta(milliseconds=100)): (source)

Store the worker-side model provider + streaming configuration.

async def backend_op(self, input: BackendOpInput) -> BackendOpOutput: (source)

Run one operation against a registered (real-I/O) backend.

async def invoke_model(self, input: ModelActivityInput) -> ModelActivityOutput: (source)

Run exactly one LLM call and return the resulting AIMessage.

async def invoke_model_streaming(self, input: ModelActivityInput) -> ModelActivityOutput: (source)

Stream one LLM call, publishing chunk batches to streaming_topic.

Token-level deltas are coalesced at streaming_batch_interval and pushed to external subscribers via the shared workflow-streams topic; the aggregated final AIMessage is returned to the workflow so the durable result is identical to the non-streaming path.

async def invoke_tool(self, input: ToolActivityInput) -> ToolActivityOutput: (source)

Execute one registered tool and return its ToolMessage.

def _build_bound_model(self, input: ModelActivityInput) -> Any: (source)

Undocumented

_model_provider = (source)

Undocumented

_streaming_batch_interval = (source)

Undocumented