enterprise_ai.runtime.temporal

Source: sdks/python/src/enterprise_ai/runtime/temporal.py. Back to the Python SDK reference index.

note

This page is generated by tools/docs/gen_sdk_reference.py from the SDK sources and the OpenAPI text folders. Do not edit it by hand: change the source and regenerate.

Functions

make_activity

make_activity(runner: AgentRunner) -> Any

Activity callable: {"messages": [...]} or {"resume": state} -> TurnResult.state().

Name	Type	Default	Required	Description
runner	AgentRunner		yes	Navin runner that executes the turn.

Returns: Any. Function payload -> dict that heartbeats when run inside a Temporal activity, then runs or resumes one turn.

drive

drive(runner: AgentRunner, task: Any = None, system: str | None = None, messages: list[dict[str, Any]] | None = None) -> TurnResult

Run the activity outside Temporal (same control flow as the workflow).

Name	Type	Default	Required	Description
runner	AgentRunner		yes	Navin runner that executes the turn.
task	Any	None	no	Task appended as the last user message.
system	str | None	None	no	System prompt inserted as the first message.
messages	list[dict[str, Any]] | None	None	no	Messages that already open the turn.

Returns: TurnResult. A TurnResult: Outcome of :meth:AgentRunner.run_turn.

build_workflow

build_workflow(runner: AgentRunner, activity_timeout: timedelta = timedelta(minutes=5)) -> tuple[type, list[Any]]

Return (AgentWorkflow, activities) ready to register on a Temporal worker.

Name	Type	Default	Required	Description
runner	AgentRunner		yes	Navin runner called by the activity.
activity_timeout	timedelta	timedelta(minutes=5)	no	Start to close and heartbeat timeout of the eai_run_turn activity (default 5 minutes).

Returns: tuple[type, list[Any]]. Tuple of the EaiAgentWorkflow class and the list holding the eai_run_turn activity; raises an error naming the temporal extra when temporalio is missing.