Add Claude Code and Pi workers

examples/workers/approvals.py

@@ -3,7 +3,7 @@
 With `approve=True`, every tool call the CLI makes becomes a `ToolApproval`
 hook. Answer it with `ai.resolve_hook`, the way any tool approval is.
 
-    uv run python examples/workers/approvals.py [codex]
+    uv run python examples/workers/approvals.py [claude-code|codex|pi]
 """
 
 import asyncio
@@ -14,29 +14,33 @@
 from ai import workers
 
 KINDS: dict[str, Callable[..., workers.Worker]] = {
+    "claude-code": workers.ClaudeCode,
     "codex": workers.Codex,
+    "pi": workers.Pi,
 }
 
 
 async def main(kind: str) -> None:
     messages = [ai.user_message("Create notes.txt containing 'hello'.")]
-    async with KINDS[kind](approve=True) as worker:
-        async with ai.Agent().run(worker, messages) as run:
-            async for event in run:
-                match event:
-                    case ai.events.PendingHookEvent(hook=hook):
-                        meta = hook.metadata
-                        answer = await asyncio.to_thread(
-                            input,
-                            f"allow {meta['tool']} {meta['kwargs']}? [y/N] ",
-                        )
-                        granted = answer.strip().lower() == "y"
-                        ai.resolve_hook(
-                            hook, {"granted": granted, "reason": "user said no"}
-                        )
-                    case ai.events.TextDelta():
-                        print(event.chunk, end="", flush=True)
-        print()
-
-
-asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "codex"))
+    async with (
+        KINDS[kind](approve=True) as worker,
+        ai.Agent().run(worker, messages) as run,
+    ):
+        async for event in run:
+            match event:
+                case ai.events.PendingHookEvent(hook=hook):
+                    meta = hook.metadata
+                    answer = await asyncio.to_thread(
+                        input,
+                        f"allow {meta['tool']} {meta['kwargs']}? [y/N] ",
+                    )
+                    granted = answer.strip().lower() == "y"
+                    ai.resolve_hook(
+                        hook, {"granted": granted, "reason": "user said no"}
+                    )
+                case ai.events.TextDelta():
+                    print(event.chunk, end="", flush=True)
+    print()
+
+
+asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "claude-code"))

examples/workers/claude_code/agent_tools.py

@@ -0,0 +1,46 @@
+"""Claude Code as the model of an `ai.Agent`: the agent's tools reach it by MCP.
+
+Claude Code calls `lookup_order`. The worker hands the call to the agent, the
+agent runs the tool, and Claude Code gets the result and goes on.
+
+    uv run python examples/workers/claude_code/agent_tools.py
+"""
+
+import asyncio
+
+import ai
+from ai import workers
+
+
+@ai.tool
+async def lookup_order(order_id: str) -> str:
+    """Look up the status of an order by its id."""
+    return f"order {order_id}: shipped on 2026-11-02"
+
+
+async def main() -> None:
+    agent = ai.Agent(tools=[lookup_order])
+    messages = [
+        ai.user_message(
+            "Use the lookup_order tool for order A-17, then tell me its "
+            "status in one sentence."
+        )
+    ]
+    async with (
+        workers.ClaudeCode() as worker,
+        agent.run(worker, messages) as run,
+    ):
+        async for event in run:
+            match event:
+                case ai.events.ToolEnd():  # the agent runs this one
+                    call = event.tool_call
+                    print(f"[agent tool] {call.tool_name} {call.tool_args}")
+                case ai.events.BuiltinToolEnd():  # Claude Code's own tool
+                    print(f"[cli tool] {event.tool_call.tool_name}")
+                case ai.events.TextDelta():
+                    print(event.chunk, end="", flush=True)
+    print()
+    print([m.role for m in run.messages])
+
+
+asyncio.run(main())

examples/workers/codex/agent_tools.py

@@ -0,0 +1,46 @@
+"""Codex as the model of an `ai.Agent`: the agent's tools reach it by MCP.
+
+Codex calls `lookup_order`. The worker hands the call to the agent, the
+agent runs the tool, and Codex gets the result and goes on.
+
+    uv run python examples/workers/codex/agent_tools.py
+"""
+
+import asyncio
+
+import ai
+from ai import workers
+
+
+@ai.tool
+async def lookup_order(order_id: str) -> str:
+    """Look up the status of an order by its id."""
+    return f"order {order_id}: shipped on 2026-11-02"
+
+
+async def main() -> None:
+    agent = ai.Agent(tools=[lookup_order])
+    messages = [
+        ai.user_message(
+            "Use the lookup_order tool for order A-17, then tell me its "
+            "status in one sentence."
+        )
+    ]
+    async with (
+        workers.Codex() as worker,
+        agent.run(worker, messages) as run,
+    ):
+        async for event in run:
+            match event:
+                case ai.events.ToolEnd():  # the agent runs this one
+                    call = event.tool_call
+                    print(f"[agent tool] {call.tool_name} {call.tool_args}")
+                case ai.events.BuiltinToolEnd():  # Codex's own tool
+                    print(f"[cli tool] {event.tool_call.tool_name}")
+                case ai.events.TextDelta():
+                    print(event.chunk, end="", flush=True)
+    print()
+    print([m.role for m in run.messages])
+
+
+asyncio.run(main())

examples/workers/pi/agent_tools.py

@@ -0,0 +1,46 @@
+"""Pi as the model of an `ai.Agent`: the agent's tools reach it by MCP.
+
+pi calls `lookup_order`. The worker hands the call to the agent, the
+agent runs the tool, and pi gets the result and goes on.
+
+    uv run python examples/workers/pi/agent_tools.py
+"""
+
+import asyncio
+
+import ai
+from ai import workers
+
+
+@ai.tool
+async def lookup_order(order_id: str) -> str:
+    """Look up the status of an order by its id."""
+    return f"order {order_id}: shipped on 2026-11-02"
+
+
+async def main() -> None:
+    agent = ai.Agent(tools=[lookup_order])
+    messages = [
+        ai.user_message(
+            "Use the lookup_order tool for order A-17, then tell me its "
+            "status in one sentence."
+        )
+    ]
+    async with (
+        workers.Pi() as worker,
+        agent.run(worker, messages) as run,
+    ):
+        async for event in run:
+            match event:
+                case ai.events.ToolEnd():  # the agent runs this one
+                    call = event.tool_call
+                    print(f"[agent tool] {call.tool_name} {call.tool_args}")
+                case ai.events.BuiltinToolEnd():  # pi's own tool
+                    print(f"[cli tool] {event.tool_call.tool_name}")
+                case ai.events.TextDelta():
+                    print(event.chunk, end="", flush=True)
+    print()
+    print([m.role for m in run.messages])
+
+
+asyncio.run(main())

examples/workers/streaming.py

@@ -1,6 +1,6 @@
 """A worker where a model goes: stream it, then continue the conversation.
 
-uv run python examples/workers/streaming.py [codex]
+uv run python examples/workers/streaming.py [claude-code|codex|pi]
 """
 
 import asyncio
@@ -11,7 +11,9 @@
 from ai import workers
 
 KINDS: dict[str, Callable[..., workers.Worker]] = {
+    "claude-code": workers.ClaudeCode,
     "codex": workers.Codex,
+    "pi": workers.Pi,
 }
 
 
@@ -49,4 +51,4 @@ async def main(kind: str) -> None:
         print(reply.text)
 
 
-asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "codex"))
+asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "claude-code"))

pyproject.toml

@@ -35,6 +35,7 @@ dependencies = [
 
 [project.optional-dependencies]
 anthropic = ["anthropic>=0.83.0,<2.0.0"]
+claude-code = ["claude-agent-sdk>=0.2.144,<0.3"]
 mcp = ["mcp>=2.0.0,<3.0.0"]
 openai = ["openai>=2.34.0,<4.0.0"]
 otel = [
@@ -69,6 +70,7 @@ bump = true
 [dependency-groups]
 dev = [
     "anthropic>=0.83.0",
+    "claude-agent-sdk>=0.2.144,<0.3",
     "mcp>=2.0.0,<3.0.0",
     "python-dotenv>=1.2.1",
     "pytest>=8.0",
@@ -82,6 +84,7 @@ dev = [
     "vercel-oidc>=0.8.0",
     "opentelemetry-sdk>=1.43.0",
     "opentelemetry-exporter-otlp-proto-http>=1.43.0",
+    "pytest-timeout>=2.3",
     "typesafe-sdk>=0.7.2",
 ]
 
@@ -115,6 +118,14 @@ exclude = [
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 testpaths = ["tests"]
+# Live worker tests drive real agent CLIs, so they are opt-in:
+# `uv run pytest -m live tests/workers` (a later -m overrides this one).
+addopts = "-m 'not live'"
+# A live test drives a real agent; hanging must be a failure.
+timeout = 300
+markers = [
+    "live: drives a real Claude Code, Codex or Pi (deselect with -m 'not live')",
+]
 filterwarnings = [
     "ignore:'asyncio\\.DefaultEventLoopPolicy' is deprecated.*:DeprecationWarning:async_solipsism\\.loop",
 ]

src/ai/providers/_optional.py

@@ -12,17 +12,25 @@
 
 
 def import_optional_sdk(
-    module_name: str, *, provider: str, extra: str
+    module_name: str,
+    *,
+    provider: str | None = None,
+    extra: str,
+    feature: str | None = None,
 ) -> ModuleType:
-    """Import an optional upstream SDK or raise a helpful installation error."""
+    """Import an optional upstream SDK or raise a helpful installation error.
+
+    ``feature`` names what needs the SDK when it is not a provider.
+    """
     root_module = module_name.partition(".")[0]
+    feature = feature or f"the {provider} provider"
     try:
         return importlib.import_module(module_name)
     except ModuleNotFoundError as exc:
         if exc.name not in {module_name, root_module}:
             raise
         raise ai_errors.InstallationError(
-            f"could not import `{root_module}`, which is required to use the "
-            f"{provider} provider, you can install it with `pip install "
+            f"could not import `{root_module}`, which is required to use "
+            f"{feature}, you can install it with `pip install "
             f'"ai[{extra}]"` or `uv add "ai[{extra}]"`'
         ) from exc

src/ai/workers/__init__.py

@@ -1,4 +1,4 @@
-"""Coding-agent CLIs as workers.
+"""Coding-agent CLIs as workers: Claude Code, Codex and Pi.
 
 Experimental: not part of the stable API, may change or be removed.
 
@@ -7,7 +7,7 @@
     import ai
     from ai import workers
 
-    async with workers.Codex(model="gpt-5.5") as worker:
+    async with workers.ClaudeCode(model="sonnet") as worker:
         async with ai.stream(worker, [ai.user_message("Fix the tests")]) as s:
             async for event in s:
                 ...
@@ -20,24 +20,28 @@
 
 A worker is a pydantic model: `worker.model_dump()` saves the conversation,
 `workers.Worker.model_validate(data)` continues it, `worker.model_copy()` is
-a new worker on the same settings, and `Codex(session_id=...)` resumes
+a new worker on the same settings, and `ClaudeCode(session_id=...)` resumes
 a conversation the CLI has stored.
 """
 
-from ._adapters.base import SessionInfo
+from ._adapter import SessionInfo
 from ._errors import (
     ExecutableMissingError,
     NotAuthenticatedError,
     TranscriptMismatchError,
     WorkerError,
 )
-from ._kinds import Codex
 from ._worker import Worker
+from .claude_code import ClaudeCode
+from .codex import Codex
+from .pi import Pi
 
 __all__ = [
+    "ClaudeCode",
     "Codex",
     "ExecutableMissingError",
     "NotAuthenticatedError",
+    "Pi",
     "SessionInfo",
     "TranscriptMismatchError",
     "Worker",