temporalio.contrib.google_genai._temporal_agents.TemporalAsyncAgents

class documentation

class TemporalAsyncAgents: (source)

Constructor: TemporalAsyncAgents(activity_config)

View In Hierarchy

Agents resource shim that routes calls through activities.

Methods accept the same keyword arguments as the real resource and forward them verbatim — the SDK validates them on the worker side, so a bad argument surfaces as an activity failure (retried per the activity's retry policy) rather than a workflow-side error.

with_raw_response / with_streaming_response are not supported in workflows.

Method __init__ Initialize with activity config for agent operation timeouts.
Async Method create Create a managed agent definition via a Temporal activity.
Async Method delete Delete a managed agent definition via a Temporal activity.
Async Method get Get a managed agent definition via a Temporal activity.
Async Method list List managed agent definitions via a Temporal activity.
Property with_raw_response Raise — raw responses are not available in workflows.
Property with_streaming_response Raise — streaming responses are not available in workflows.
Method _config Undocumented
Instance Variable _activity_config Undocumented
def __init__(self, activity_config: ActivityConfig | None = None): (source)

Initialize with activity config for agent operation timeouts.

async def create(self, **kwargs: Any) -> Agent: (source)

Create a managed agent definition via a Temporal activity.

async def delete(self, id: str, **kwargs: Any) -> AgentDeleteResponse: (source)

Delete a managed agent definition via a Temporal activity.

async def get(self, id: str, **kwargs: Any) -> Agent: (source)

Get a managed agent definition via a Temporal activity.

async def list(self, **kwargs: Any) -> AgentListResponse: (source)

List managed agent definitions via a Temporal activity.

with_raw_response: Any = (source)

Raise — raw responses are not available in workflows.

with_streaming_response: Any = (source)

Raise — streaming responses are not available in workflows.

def _config(self, summary: str, params: dict[str, Any]) -> ActivityConfig: (source)

Undocumented

_activity_config = (source)

Undocumented