Let Agent be run in a Temporal workflow by moving model requests, tool calls, and MCP to Temporal activities

[DouweM]

Closes #1975

Run it yourself:

• brew install temporal
• temporal server start-dev
• uv run temporal.py

To do:

• google-genai + temporalio type hinting conflict
• Ensure Logfire works
• Use Temporal Plugins
• Let TemporalSettings be set for a specific toolset
• Error on un(de)serializable deps
• Use temporalio package instead of Git repo once Plugin Support temporalio/sdk-python#952 is released
• Split out Agent(event_stream_handler=...), merge that first Add Agent event_stream_handler function #2306
• Spit out Toolset.id, merge that first
• Tests https://github.com/temporalio/sdk-python/blob/3244f8bffebee05e0e7efefb1240a75039903dda/tests/conftest.py#L100
• Docs

[github-actions]

Docs Preview

commit:	090ec23
Preview URL:	https://95077b1b-pydantic-ai-previews.pydantic.workers.dev

[mattbrandman]

I'm so excited for this to be natively supported

[nir-litt]

Thank you for taking the time to look into native temporal support - this is a topic I’m really interested in seeing progress on.

One key feature I’d love to see is support for HITL (Human-in-the-Loop) during tool calls when using temporal and agents. My suggestion on how to achieve that (and not being to specific for HITL) is to pause the workflow before a tool call waiting on a condition for a signal. This will allow, among other things, to wait for human input.

Currently, tool calls are handled like this:

async def call_tool(
    self, name: str, tool_args: dict[str, Any], ctx: RunContext[AgentDepsT], tool: ToolsetTool[AgentDepsT]
) -> Any:
    serialized_run_context = self.serialize_run_context(ctx)
    return await workflow.execute_activity(
        activity=self.call_tool_activity,
        arg=FunctionCallToolParams(name=name, tool_args=tool_args, serialized_run_context=serialized_run_context),
        **self.temporal_settings.__dict__,
    )

To add the wait support, we can do something like:
(This example is a bit abstract - we will have to understand (at least):

• how to pass the self.allow_tool_call from the workflow to this class or effect it with some callback instead of just setting it
• pass all the data of the tool call for the handler to be able to decide which action to take)

async def call_tool(
    self, name: str, tool_args: dict[str, Any], ctx: RunContext[AgentDepsT], tool: ToolsetTool[AgentDepsT]
) -> Any:
    # --- ADDED: tool call pause using signal and wait_condition ---
    await workflow.wait_condition(lambda: self.allow_tool_call)
    # --- END ADDED CODE ---
    serialized_run_context = self.serialize_run_context(ctx)
    activity_result =  await workflow.execute_activity(
        activity=self.call_tool_activity,
        arg=FunctionCallToolParams(name=name, tool_args=tool_args, serialized_run_context=serialized_run_context),
        **self.temporal_settings.__dict__,
    )
    # --- ADDED: tool call reset state ---
    self.allow_tool_call = False
    return activity_result
    # --- END ADDED CODE ---

This will allow users to define their workflow something like:

@workflow.signal
def approve_tool_call(self):
    ... # some code to determine the tool is OK to run
    self.allow_tool_call = True

This way, the workflow will pause at the tool call and only continue once the signal is approved.

of course this would be optional and the default ,that can be overwritten by the user, will be to approve the request

Appreciate the ongoing work!

[hyperlint-ai]

PR Change Summary

Introduced the Temporal Agent feature by enhancing toolset definitions and adding optional IDs for better identification in error messages.

• Added optional ID parameters to FunctionToolset for better identification in error messages.
• Updated LangChainToolset and ACIToolset to include IDs for Slack and Open Weather Map respectively.
• Enhanced documentation to clarify the use of IDs in durable execution environments.

Modified Files

• docs/tools.md
• docs/toolsets.md

---

How can I customize these reviews?

Check out the Hyperlint AI Reviewer docs for more information on how to customize the review.

If you just want to ignore it on this PR, you can add the hyperlint-ignore label to the PR. Future changes won't trigger a Hyperlint review.

Note specifically for link checks, we only check the first 30 links in a file and we cache the results for several hours (for instance, if you just added a page, you might experience this). Our recommendation is to add hyperlint-ignore to the PR to ignore the link check for this PR.

[DouweM]

Consider disabling retries within an activity, and having it all be handled by Temporal, which will have to know whether an error is retryable and how long to back off for: https://github.com/temporalio/sdk-python/blob/3244f8bffebee05e0e7efefb1240a75039903dda/temporalio/contrib/openai_agents/_invoke_model_activity.py#L231

[DouweM]

Either way we should document the 3 levels of retries: Temporal > https://ai.pydantic.dev/retries/ > provider client

[DouweM]

To do:

• override run_stream and iter to point the user to run/run_sync + event_stream_handler