Implement ai.workers

examples/workers/agent_tools.py

@@ -0,0 +1,50 @@
+"""A worker as the model of an `ai.Agent`: the agent's tools reach it by MCP.
+
+The CLI calls `lookup_order`. The worker hands the call to the agent, the
+agent runs the tool, and the CLI gets the result and goes on.
+
+    uv run python examples/workers/agent_tools.py [codex]
+"""
+
+import asyncio
+import sys
+from collections.abc import Callable
+
+import ai
+from ai import workers
+
+KINDS: dict[str, Callable[..., workers.Worker]] = {
+    "codex": workers.Codex,
+}
+
+
+@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(kind: str) -> 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 KINDS[kind]() as worker:
+        async with 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():  # the CLI'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(sys.argv[1] if len(sys.argv) > 1 else "codex"))

examples/workers/approvals.py

@@ -0,0 +1,42 @@
+"""Approve each of the CLI's own tool calls, as `ai.Agent` tool approvals.
+
+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]
+"""
+
+import asyncio
+import sys
+from collections.abc import Callable
+
+import ai
+from ai import workers
+
+KINDS: dict[str, Callable[..., workers.Worker]] = {
+    "codex": workers.Codex,
+}
+
+
+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"))

examples/workers/streaming.py

@@ -0,0 +1,52 @@
+"""A worker where a model goes: stream it, then continue the conversation.
+
+uv run python examples/workers/streaming.py [codex]
+"""
+
+import asyncio
+import sys
+from collections.abc import Callable
+
+import ai
+from ai import workers
+
+KINDS: dict[str, Callable[..., workers.Worker]] = {
+    "codex": workers.Codex,
+}
+
+
+async def main(kind: str) -> None:
+    async with KINDS[kind]() as worker:
+        messages = [
+            ai.user_message(
+                "List the files here with a shell command, then say DONE."
+            )
+        ]
+        async with ai.stream(worker, messages) as s:
+            async for event in s:
+                match event:
+                    case ai.events.TextDelta():
+                        print(event.chunk, end="", flush=True)
+                    case ai.events.BuiltinToolEnd():  # the CLI's own tool
+                        call = event.tool_call
+                        print(f"\n[{call.tool_name}] {call.tool_args[:60]}")
+        print(f"\nusage: {s.message.usage}")
+
+        # The worker keeps the conversation: send it back with what is new.
+        follow_up = [
+            *worker.messages,
+            ai.user_message("How many files? Reply with a number only."),
+        ]
+        print(await ai.experimental_generate(worker, follow_up))
+
+    # A worker is data: save it, and continue the conversation later.
+    saved = worker.model_dump_json()
+    again = workers.Worker.model_validate_json(saved)
+    async with again:
+        reply = await ai.experimental_generate(
+            again, [*again.messages, ai.user_message("And the largest one?")]
+        )
+        print(reply.text)
+
+
+asyncio.run(main(sys.argv[1] if len(sys.argv) > 1 else "codex"))

pyproject.toml

@@ -176,6 +176,11 @@ extend-ignore = [
     "TC003",   # annoying and wrong if TYPE_CHECKING for stdlib imports
 ]
 
+[tool.ruff.lint.flake8-type-checking]
+# pydantic evaluates field annotations at runtime, so their imports must stay
+# real even under `from __future__ import annotations`
+runtime-evaluated-base-classes = ["pydantic.BaseModel"]
+
 [tool.ruff.lint.per-file-ignores]
 "examples/**/*.py" = [
     "D",       # examples are executable samples, not public API

src/ai/__init__.py

@@ -8,6 +8,7 @@
     testing,
     ui,
     util,
+    workers,
 )
 from .agents import (
     Agent,
@@ -229,5 +230,6 @@
     "ui",
     "user_message",
     "util",
+    "workers",
     "yield_from",
 ]

src/ai/agents/agent.py

@@ -1040,11 +1040,16 @@ async def _iterate(self) -> AsyncGenerator[events_.ToolCallResult]:
 
 
 def _validate_model(value: Any) -> models.ModelLike:
-    # Another model-like is kept as it is; a dict from JSON is an ai.Model.
+    # Another model-like (an ai.workers.Worker) is kept as it is. A dict
+    # from JSON with a worker's `kind` is a worker; any other is an ai.Model.
     if isinstance(value, models.ModelLike) and not isinstance(
         value, models.Model
     ):
         return value
+    if isinstance(value, dict) and "kind" in value:
+        from .. import workers  # noqa: PLC0415
+
+        return workers.Worker.model_validate(value)
     return models.Model.model_validate(value)
 
 

src/ai/workers/__init__.py

@@ -0,0 +1,45 @@
+"""Coding-agent CLIs as workers.
+
+Experimental: not part of the stable API, may change or be removed.
+
+A worker goes where a model goes, and it keeps its conversation::
+
+    import ai
+    from ai import workers
+
+    async with workers.Codex(model="gpt-5.5") as worker:
+        async with ai.stream(worker, [ai.user_message("Fix the tests")]) as s:
+            async for event in s:
+                ...
+
+        # an ai.Agent's tools reach the CLI through MCP
+        async with ai.Agent(tools=[search_docs]).run(worker, messages) as run:
+            ...
+
+The CLI's own tools come back as `ai.events.BuiltinTool*` events.
+
+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 conversation the CLI has stored.
+"""
+
+from ._adapters.base import SessionInfo
+from ._errors import (
+    ExecutableMissingError,
+    NotAuthenticatedError,
+    TranscriptMismatchError,
+    WorkerError,
+)
+from ._kinds import Codex
+from ._worker import Worker
+
+__all__ = [
+    "Codex",
+    "ExecutableMissingError",
+    "NotAuthenticatedError",
+    "SessionInfo",
+    "TranscriptMismatchError",
+    "Worker",
+    "WorkerError",
+]