Add Pydantic AI plugin samples

README.md

@@ -101,6 +101,7 @@ without wrapping them in a workflow.
 * [patching](patching) - Alter workflows safely with `patch` and `deprecate_patch`.
 * [polling](polling) - Recommended implementation of an activity that needs to periodically poll an external resource waiting its successful completion.
 * [prometheus](prometheus) - Configure Prometheus metrics on clients/workers.
+* [pydantic_ai_plugin](pydantic_ai_plugin) - Run Pydantic AI agents as durable Temporal workflows, including chat, tools, MCP, HITL, structured output, streaming, multi-agent orchestration, and Logfire.
 * [pydantic_converter](pydantic_converter) - Data converter for using Pydantic models.
 * [pydantic_converter_v1](pydantic_converter_v1) - Data converter for Pydantic v1 models (prefer pydantic_converter for v2).
 * [replay](replay) - Verify that workflow code changes are compatible with existing histories.

pydantic_ai_plugin/README.md

@@ -0,0 +1,35 @@
+# Pydantic AI Plugin Samples
+
+These samples run ordinary [Pydantic AI](https://ai.pydantic.dev/) agents inside Temporal Workflows with `TemporalDurability`. `PydanticAIPlugin` configures Pydantic payload conversion, the Workflow Sandbox, non-retryable failures, and automatic Activity registration for agents declared on `PydanticAIWorkflow` classes.
+
+The samples use Pydantic AI's credential-free `TestModel` by default. Pass an OpenAI model name in the Workflow input to use a live model instead.
+
+| Sample | What it demonstrates |
+| --- | --- |
+| [chat](chat) | Multi-turn Updates, durable message history, and Continue-As-New. |
+| [tools](tools) | A deterministic in-Workflow tool beside an I/O-style Activity tool. |
+| [mcp](mcp) | A stateless stdio MCP server whose operations run as Activities. |
+| [streaming](streaming) | Pydantic AI events over Temporal Workflow Streams. |
+| [human_in_the_loop](human_in_the_loop) | Approval, rejection, and cancellation decisions delivered by Signal. |
+| [structured_output](structured_output) | A typed Pydantic model crossing Activity and Workflow boundaries. |
+| [multi_agent](multi_agent) | Durable researcher and writer agents coordinated by one Workflow. |
+| [logfire](logfire) | Pydantic AI and Temporal tracing wired together by `LogfirePlugin`. |
+
+## Install
+
+```bash
+uv sync --group pydantic-ai
+temporal server start-dev
+```
+
+The dependency group pins `pydantic-ai-slim` to commit `2b45faa97e76461c60500e9755a130b158a2418d`, the head of [pydantic-ai PR #6639](https://github.com/pydantic/pydantic-ai/pull/6639), because the Workflow Streams API used by the streaming sample is not yet released.
+
+## Run
+
+Start a category's Worker, then use the Temporal CLI command in that category's README. With no `model` value, the Workflow uses `TestModel` and needs no API key or network access. For a live run, set `model` to `gateway/openai:gpt-5.2` and export `PYDANTIC_AI_GATEWAY_API_KEY` in the Worker process.
+
+The examples use the declarative registration path: each Workflow subclasses `PydanticAIWorkflow`, lists its agents in `__pydantic_ai_agents__`, and the Client installs `PydanticAIPlugin`. `AgentPlugin(agent)` is the narrower Worker-only alternative when a Workflow class cannot declare its agents; it does not replace the Client-side `PydanticAIPlugin` configuration.
+
+## Unsupported
+
+**Sandboxes:** Pydantic AI does not provide an agent-facing isolation environment whose sessions and shell or filesystem operations are durably managed by this integration.

pydantic_ai_plugin/__init__.py

@@ -0,0 +1 @@
+"""Pydantic AI integration samples."""

pydantic_ai_plugin/chat/README.md

@@ -0,0 +1,23 @@
+# Agent chat
+
+`ChatWorkflow` accepts each user turn as an Update, returns the assistant response from that Update, and keeps Pydantic AI `ModelMessage` history in Workflow state. When Temporal recommends Continue-As-New, it waits for active handlers and carries the typed history into the next run.
+
+Assign each independent chat a unique Workflow ID. Continue-As-New keeps that ID across the whole chain automatically, so do not reuse it for an unrelated workflow.
+
+```bash
+uv run pydantic_ai_plugin/chat/run_worker.py
+
+WORKFLOW_ID="pydantic-ai-chat-$(uv run python -c 'import uuid; print(uuid.uuid4())')"
+temporal workflow start --type ChatWorkflow --task-queue pydantic-ai-chat --workflow-id "$WORKFLOW_ID" --input '{"messages":[],"model":null}'
+temporal workflow update --workflow-id "$WORKFLOW_ID" --name turn --input '"Hello"'
+temporal workflow query --workflow-id "$WORKFLOW_ID" --type message_count
+temporal workflow signal --workflow-id "$WORKFLOW_ID" --name end_chat
+```
+
+For a live model, start with `{"messages":[],"model":"gateway/openai:gpt-5.2"}`.
+
+Expected offline Update result: `Ready for the next turn.`
+
+```bash
+uv run --group pydantic-ai pytest tests/pydantic_ai_plugin/chat_test.py
+```

pydantic_ai_plugin/chat/__init__.py

@@ -0,0 +1 @@
+"""Durable chat sample."""

pydantic_ai_plugin/chat/run_worker.py

@@ -0,0 +1,24 @@
+import asyncio
+import os
+
+from pydantic_ai.durable_exec.temporal import PydanticAIPlugin
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from pydantic_ai_plugin.chat.workflow import ChatWorkflow
+
+
+async def main() -> None:
+    client = await Client.connect(
+        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
+        plugins=[PydanticAIPlugin()],
+    )
+    await Worker(
+        client,
+        task_queue="pydantic-ai-chat",
+        workflows=[ChatWorkflow],
+    ).run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pydantic_ai_plugin/chat/workflow.py

@@ -0,0 +1,67 @@
+import asyncio
+from dataclasses import dataclass, field
+
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from pydantic_ai import Agent, ModelMessage
+    from pydantic_ai.durable_exec.temporal import (
+        PydanticAIWorkflow,
+        TemporalDurability,
+    )
+    from pydantic_ai.models.test import TestModel
+
+
+@dataclass
+class ChatInput:
+    messages: list[ModelMessage] = field(default_factory=list)
+    model: str | None = None
+
+
+agent = Agent(
+    TestModel(custom_output_text="Ready for the next turn."),
+    name="durable_chat",
+    capabilities=[TemporalDurability()],
+)
+
+
+@workflow.defn
+class ChatWorkflow(PydanticAIWorkflow):
+    __pydantic_ai_agents__ = [agent]
+
+    @workflow.init
+    def __init__(self, input: ChatInput) -> None:
+        self._input = input
+        self._messages = list(input.messages)
+        self._done = False
+        self._lock = asyncio.Lock()
+
+    @workflow.update
+    async def turn(self, prompt: str) -> str:
+        async with self._lock:
+            result = await agent.run(
+                prompt,
+                message_history=self._messages,
+                model=self._input.model,
+            )
+            self._messages = result.all_messages()
+            return result.output
+
+    @workflow.signal
+    def end_chat(self) -> None:
+        self._done = True
+
+    @workflow.query
+    def message_count(self) -> int:
+        return len(self._messages)
+
+    @workflow.run
+    async def run(self, input: ChatInput) -> None:
+        await workflow.wait_condition(
+            lambda: self._done or workflow.info().is_continue_as_new_suggested()
+        )
+        await workflow.wait_condition(workflow.all_handlers_finished)
+        if not self._done:
+            workflow.continue_as_new(
+                ChatInput(messages=self._messages, model=input.model)
+            )

pydantic_ai_plugin/human_in_the_loop/README.md

@@ -0,0 +1,22 @@
+# Human in the loop
+
+The `delete_record` tool requires approval. Pydantic AI ends the first run with `DeferredToolRequests`; the Workflow exposes the pending tool through a Query and waits durably for an `approve`, `reject`, or `cancel` Signal.
+
+Approval resumes the agent with `DeferredToolResults`. Rejection sends `ToolDenied` back to the model. Cancellation ends the business operation without executing the tool.
+
+```bash
+uv run pydantic_ai_plugin/human_in_the_loop/run_worker.py
+
+temporal workflow start --type ApprovalWorkflow --task-queue pydantic-ai-approval --workflow-id pydantic-ai-approval-1 --input '{"prompt":"Delete record 42.","model":null}'
+temporal workflow query --workflow-id pydantic-ai-approval-1 --type pending_approval
+temporal workflow signal --workflow-id pydantic-ai-approval-1 --name decide --input '"approve"'
+temporal workflow show --workflow-id pydantic-ai-approval-1
+```
+
+Send `"reject"` or `"cancel"` to exercise the other paths. Use `"model":"gateway/openai:gpt-5.2"` for a live run.
+
+Expected offline result after approval or rejection: `The operator decision was applied.` Cancellation returns `Cancelled by the operator.`
+
+```bash
+uv run --group pydantic-ai pytest tests/pydantic_ai_plugin/human_in_the_loop_test.py
+```

pydantic_ai_plugin/human_in_the_loop/__init__.py

@@ -0,0 +1 @@
+"""Human-in-the-loop sample."""

pydantic_ai_plugin/human_in_the_loop/run_worker.py

@@ -0,0 +1,24 @@
+import asyncio
+import os
+
+from pydantic_ai.durable_exec.temporal import PydanticAIPlugin
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from pydantic_ai_plugin.human_in_the_loop.workflow import ApprovalWorkflow
+
+
+async def main() -> None:
+    client = await Client.connect(
+        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
+        plugins=[PydanticAIPlugin()],
+    )
+    await Worker(
+        client,
+        task_queue="pydantic-ai-approval",
+        workflows=[ApprovalWorkflow],
+    ).run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pydantic_ai_plugin/human_in_the_loop/workflow.py

@@ -0,0 +1,92 @@
+from dataclasses import dataclass
+from typing import Literal
+
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from pydantic_ai import (
+        Agent,
+        DeferredToolRequests,
+        ToolApproved,
+        ToolDenied,
+    )
+    from pydantic_ai.durable_exec.temporal import (
+        PydanticAIWorkflow,
+        TemporalDurability,
+    )
+    from pydantic_ai.models.test import TestModel
+
+
+Decision = Literal["approve", "reject", "cancel"]
+
+
+@dataclass
+class ApprovalInput:
+    prompt: str
+    model: str | None = None
+
+
+agent = Agent(
+    TestModel(
+        call_tools=["delete_record"],
+        custom_output_text="The operator decision was applied.",
+    ),
+    name="approval_agent",
+    output_type=[str, DeferredToolRequests],
+    capabilities=[TemporalDurability()],
+)
+
+
+@agent.tool_plain(requires_approval=True)
+async def delete_record(record_id: str) -> str:
+    return f"deleted {record_id}"
+
+
+@workflow.defn
+class ApprovalWorkflow(PydanticAIWorkflow):
+    __pydantic_ai_agents__ = [agent]
+
+    def __init__(self) -> None:
+        self._decision: Decision | None = None
+        self._pending: str | None = None
+
+    @workflow.signal
+    def decide(self, decision: Decision) -> None:
+        self._decision = decision
+
+    @workflow.query
+    def pending_approval(self) -> str | None:
+        return self._pending
+
+    @workflow.run
+    async def run(self, input: ApprovalInput) -> str:
+        first = await agent.run(input.prompt, model=input.model)
+        if not isinstance(first.output, DeferredToolRequests):
+            return first.output
+
+        calls = first.output.approvals
+        self._pending = calls[0].tool_name if calls else None
+        await workflow.wait_condition(lambda: self._decision is not None)
+        decision = self._decision
+        self._pending = None
+
+        if decision == "cancel":
+            return "Cancelled by the operator."
+
+        approvals: dict[str, bool | ToolApproved | ToolDenied] = {
+            call.tool_call_id: (
+                True
+                if decision == "approve"
+                else ToolDenied(message="Rejected by the operator.")
+            )
+            for call in calls
+        }
+        deferred_results = first.output.build_results(approvals=approvals)
+        resumed = await agent.run(
+            "Continue after the operator decision.",
+            message_history=first.all_messages(),
+            deferred_tool_results=deferred_results,
+            model=input.model,
+        )
+        assert isinstance(resumed.output, str)
+        return resumed.output

pydantic_ai_plugin/logfire/README.md

@@ -0,0 +1,16 @@
+# Observability with Logfire
+
+`LogfirePlugin` installs Temporal's OpenTelemetry tracing interceptor and Pydantic AI instrumentation, producing one connected trace across Workflow execution, model Activities, agent runs, and tools. The sample disables remote export when `LOGFIRE_TOKEN` is absent, so it also runs credential-free.
+
+```bash
+uv run pydantic_ai_plugin/logfire/run_worker.py
+temporal workflow execute --type ObservabilityWorkflow --task-queue pydantic-ai-logfire --workflow-id pydantic-ai-logfire-1 --input '{"prompt":"Run an observable agent.","model":null}'
+```
+
+To send traces to Logfire, authenticate with `logfire auth`, export `LOGFIRE_TOKEN`, and restart the Worker. Use `"model":"gateway/openai:gpt-5.2"` for a live model.
+
+Expected offline result: `This run is traced.` The test captures both the agent and Workflow spans in memory.
+
+```bash
+uv run --group pydantic-ai pytest tests/pydantic_ai_plugin/logfire_test.py
+```

pydantic_ai_plugin/logfire/__init__.py

@@ -0,0 +1 @@
+"""Logfire observability sample."""

pydantic_ai_plugin/logfire/run_worker.py

@@ -0,0 +1,27 @@
+import asyncio
+import os
+
+import logfire
+from pydantic_ai.durable_exec.temporal import LogfirePlugin, PydanticAIPlugin
+from temporalio.client import Client
+from temporalio.worker import Worker
+
+from pydantic_ai_plugin.logfire.workflow import ObservabilityWorkflow
+
+
+async def main() -> None:
+    if os.environ.get("LOGFIRE_TOKEN") is None:
+        logfire.configure(send_to_logfire=False)
+    client = await Client.connect(
+        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
+        plugins=[PydanticAIPlugin(), LogfirePlugin(metrics=False)],
+    )
+    await Worker(
+        client,
+        task_queue="pydantic-ai-logfire",
+        workflows=[ObservabilityWorkflow],
+    ).run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pydantic_ai_plugin/logfire/workflow.py

@@ -0,0 +1,34 @@
+from dataclasses import dataclass
+
+from temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from pydantic_ai import Agent
+    from pydantic_ai.durable_exec.temporal import (
+        PydanticAIWorkflow,
+        TemporalDurability,
+    )
+    from pydantic_ai.models.test import TestModel
+
+
+@dataclass
+class ObservabilityInput:
+    prompt: str
+    model: str | None = None
+
+
+agent = Agent(
+    TestModel(custom_output_text="This run is traced."),
+    name="observable_agent",
+    capabilities=[TemporalDurability()],
+)
+
+
+@workflow.defn
+class ObservabilityWorkflow(PydanticAIWorkflow):
+    __pydantic_ai_agents__ = [agent]
+
+    @workflow.run
+    async def run(self, input: ObservabilityInput) -> str:
+        result = await agent.run(input.prompt, model=input.model)
+        return result.output