# What's durable, and what isn't > Where the journal draws the line between compatibility and durability, and how to move work across it. Source: https://docs.pyrula.io/agents/integrations/durable-vs-not/ An integration is durable only when its work is recorded in the journal. API compatibility alone gives no replay or crash recovery. | Tier | Execution | Replay | |---|---|---| | **T1 durable-native** | A Pyrula run, step or tool | Recorded work replays without re-executing and may suspend for human input. | | **T3 compatibility** | A compatible API outside a run | Calls execute normally and are not journaled. | The [guarantee matrix](https://docs.pyrula.io/agents/integrations/guarantees.md) lists each capability's tier and tested SDK versions. ## Moving work from T3 to T1 - `ctx.step` around anything that charges a card, writes to a system of record, or must not run twice. - `@tool` for consequential tools, so a crash mid-agent doesn't re-fire them. - `as_langgraph_node` to make one node of a graph a Pyrula run ([LangGraph](https://docs.pyrula.io/agents/integrations/langgraph.md)). ## durable_chat_model Wraps a LangChain chat model so its async calls (`ainvoke`/`astream`) are journaled inside a Pyrula run. Install with `pip install "pyrula[agents,langchain]"`. ```python from langchain_openai import ChatOpenAI from pyrula.agents import agent from pyrula.agents.integrations import durable_chat_model model = durable_chat_model(ChatOpenAI(model="gpt-6-luna")) @agent async def my_agent(ctx, message: str) -> str: result = await model.ainvoke([{"role": "user", "content": message}]) return result.content ``` On replay the response comes from the journal. The sync path (`invoke`) raises `UnjournaledSurface` inside a run; outside a run both paths pass through. Each call is a cancel and pause boundary. Don't wrap the model in `with_fallbacks` or `with_retry`: stops arrive as exceptions, and a fallback would call a model live during replay. Use LangGraph's `RetryPolicy` on the calling node. Whole-graph behavior is on [LangGraph](https://docs.pyrula.io/agents/integrations/langgraph.md). ## The OpenAI drop-in `pyrula.agents.compat.openai.AsyncOpenAI` has the `openai.AsyncOpenAI` shape. Swap the import (`pip install "pyrula[agents]"`): ```python # from openai import AsyncOpenAI from pyrula.agents.compat.openai import AsyncOpenAI client = AsyncOpenAI() resp = await client.chat.completions.create(model="gpt-6-luna", messages=[...]) ``` Inside a run, `chat.completions.create` and `responses.create` are journaled and replay without calling the provider. Outside a run they pass through with a one-time warning. For sync code, see [Sync code and threads](#sync-code-and-threads). | Inside a run | Surfaces | |---|---| | Journaled | `chat.completions.create` and `responses.create`, including `stream=True`, `response_format=` and `text={"format": ...}`, and the SDK helpers built on them: `chat.completions.parse` / `stream` and `responses.parse` / `stream` | | Passed through | `embeddings`, `files`, `models`, `responses.delete`, `responses.input_items`, `responses.input_tokens`, `beta.assistants` and other non-generation surfaces | | Refused (would call the provider live on every replay) | `completions`, `audio`, `images`, `videos`, `realtime`, `beta.chat`, `beta.realtime`, `beta.threads`, `responses.stream(response_id=...)`, `responses.retrieve`, `responses.cancel`, `responses.compact`, `responses.connect`, and `with_raw_response` / `with_streaming_response` at the top level, under `chat`, `chat.completions`, `responses` and `beta` | Use `create(...)` or the helpers above instead of the refused surfaces; `beta.chat.completions.parse` is `chat.completions.parse`. The OpenAI Agents SDK runs on this client: see [OpenAI Agents SDK](https://docs.pyrula.io/agents/integrations/openai-agents-sdk.md) for what replays and what doesn't. #### The SDK's parse and stream helpers A helper is its `create` plus the SDK's own result builder, so it shares `create`'s journal entry: `chat.completions.parse(response_format=Model)` and `chat.completions.create(response_format=)` journal the same call and replay each other. On replay the SDK's own parser rebuilds the typed object (`message.parsed`, `output_parsed`) from the stored reply and the helper makes no provider request; the stream helpers return the SDK's own manager (`get_final_completion()`, `text_stream`, events) over the journaled stream. The request fingerprint covers the schema, so a changed `response_format` / `text_format` class refuses the resume. - A reply the parser rejects (`LengthFinishReasonError`, a pydantic `ValidationError`) is journaled as a successful call and the same error is raised again on replay. - Your model's validators and `default_factory` run again on replay, so keep them free of side effects. `request_id` is `None` on a replayed object (response headers are not stored). - A manager you build and never enter sends and journals nothing. A stream you stop early, cancel or drop is metered once as an estimate, as for `create(stream=True)`. One still open as the run parks on an interrupt, a sleep or a signal wait, or ends, is metered the same way once the calls in flight have drained (a stream that ends meanwhile is journaled exact), ahead of the question the run re-journals last; the resumed attempt's own call is metered on its own. - `responses.parse(parse_as=...)` (openai 3.42 and later) parses with the class you name, in a run or outside one, and an input whose `configuration_update` changes the format without `parse_as` raises the SDK's `TypeError` before anything is sent. `responses.stream` has no `parse_as`. - `responses.stream(response_id=...)` and `starting_after` re-read stored server state and are refused in a run. `with_raw_response` and `with_streaming_response` stay refused, `.parse` included. - If a future openai release moves the SDK internals a helper needs, the helper is refused in a run with an error naming the openai version (outside a run it calls the SDK's own helper), until the drop-in is updated. ### Chat completions and streams `chat.completions.create` returns the SDK's `ChatCompletion`, rebuilt from the journal on replay. `stream=True` returns one object that supports `async for`, `async with` and `close()`, live and on replay. It is not an `openai.AsyncStream`: its `.response` is a small stand-in (`close()`, `headers`) for the SDK's stream helpers, not the HTTP response. A streamed call is journaled, and its usage metered, when its last choice's `finish_reason` chunk arrives (with `n` above 1, once every choice has one). With usage requested (`stream_options={"include_usage": True}`, or by the drop-in inside an `@agent`), the usage chunk that follows is recorded with it. - Breaking out of the loop on the `finish_reason` chunk leaves the call journaled; a resume sends no request. - A stream you stop reading earlier is not journaled and runs live again. - A stream the provider ends without a `finish_reason` chunk, read to its end, raises `ProviderStreamTruncated`, journaled like a provider error. ### The Responses API `responses.create` is journaled like `chat.completions.create`. A replay returns the SDK's `Response` rebuilt from the journal. With `stream=True` the object supports `async for`, `async with` and `close()`, and a replay yields the recorded events without calling the provider. It is not an `openai.AsyncStream`: code that reads headers or `x-request-id` off the stream raises `AttributeError` inside a run. A streamed call is journaled when its terminal event arrives (`response.completed`, `response.incomplete`, `response.failed` or `error`). A stream you stop before it runs live again. A stream the provider ends without a terminal event raises `ProviderStreamTruncated`. Import it, and `UnjournaledSurface`, from `pyrula.agents`. The streamed and non-streamed forms are separate steps, so switching `stream` on a resumed call runs it live. ```python import asyncio import json import httpx from pyrula.agents import AgentContext, agent from pyrula.agents.compat.openai import AsyncOpenAI from pyrula.agents.testing.runner import AgentRunner provider_calls = [] def provider(request: httpx.Request) -> httpx.Response: # A stand-in for the OpenAI API, so this block runs offline. provider_calls.append(json.loads(request.content)) return httpx.Response(200, json={ "id": "resp_1", "object": "response", "created_at": 0, "model": "gpt-6-luna", "status": "completed", "parallel_tool_calls": True, "tool_choice": "auto", "tools": [], "output": [{"id": "msg_1", "type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "Pong.", "annotations": []}]}], "usage": {"input_tokens": 9, "input_tokens_details": {"cached_tokens": 0}, "output_tokens": 2, "output_tokens_details": {"reasoning_tokens": 0}, "total_tokens": 11}, }) @agent(name="responses_ping") async def ping(ctx: AgentContext) -> str: client = AsyncOpenAI( api_key="unused", base_url="http://openai.test/v1", http_client=httpx.AsyncClient(transport=httpx.MockTransport(provider)), ) response = await client.responses.create(model="gpt-6-luna", input="Ping?") return response.output_text events = asyncio.run(AgentRunner(ping).run()) assert events[-1].payload["result"] == "Pong." and len(provider_calls) == 1 ``` **Failures.** A provider error (a 429, a 404, a dropped connection) is journaled as the call's outcome and re-raised. A body that catches it and retries takes the same branch on replay, with no request sent. The rebuilt error is the SDK's own class with what code branches on. OpenAI and Anthropic: `status_code`, `request_id`, `body`, `code`, `type` and `param` (where the SDK has them), `request`, and a `response` with the status, the `retry-after`, `retry-after-ms`, `x-should-retry` and request-id headers and the provider's JSON error body. google-genai: `code`, `status`, `message`, `details` and a `response` of the same shape. The response is rebuilt, not the live one: no other header is kept, `request` is the method and URL without query or credentials, and the request's headers and body are never journaled. The body is scrubbed of secret-shaped values on the way in, and one over 16 KiB is not kept (the attribute is `None` and the response has an empty body). The `response` is an `httpx` (`httpx2` on openai 3 and anthropic 2) response of the replaying SDK's flavour, also for a google-genai error that was an aiohttp response live. `message` and `str(error)` are the live text, unscrubbed, as before. An error journaled by an earlier release replays as before, with the class and message only. **Refused arguments.** These raise `UnjournaledSurface` before anything is written or sent: - `background=True`. - `conversation=`: it is mutable server state. Chain with `previous_response_id`, or send the full `input`. - Tools that OpenAI's servers run with side effects: `mcp`, `code_interpreter`, a `shell` with a container environment, `programmatic_tool_calling`, `image_generation` and any tool type this release does not know. `function`, `custom`, `computer`, `local_shell`, `apply_patch`, a local `shell` and the read-only `web_search*`, `file_search` and `tool_search` are allowed. **Tool calls.** `function_call` items are part of the recorded response, so their `call_id`s replay unchanged. Wrap the tool's side effects in `ctx.step` or `@tool`. The next call's `function_call_output` is part of its fingerprint, so a tool that returns something different on resume raises `ReplayInputChanged`. **Server-side state.** The first live call after the replayed prefix carries the `previous_response_id` recorded earlier. That response must still be stored at OpenAI (not created with `store=False`, not deleted, within the retention window), or the provider rejects the call with a 4xx. A stored `prompt` without a pinned `version` resolves to the latest version on a live re-run. **At-least-once.** A call in flight when the process died has no recording, so a resume sends it again and the provider may bill it twice. **Replay check.** A resumed call is compared on `input`, `instructions`, `tools`, `tool_choice`, `text.format`, `previous_response_id` and the stored `prompt` (id, version, variables). Model, sampling, `store`, `metadata`, `reasoning`, `include` and `text.verbosity` are not compared. **Metering.** A Responses call inside an `@agent` counts toward `max_llm_calls`, `max_total_tokens` and `max_cost_usd`. Hosted-tool fees are not metered, so a cost cap undercounts runs that use them. With a stored `prompt` and no `model=`, the call is unpriced. **Size.** Each journaled event is capped at 1,000,000 bytes by default. A streamed call records every event, roughly 170 bytes per five-character delta. A longer streamed answer ends the run with `PayloadTooLarge` after the call was paid for. Raise `max_event_payload_bytes` or stream shorter answers. ### Calls inside a step or tool A drop-in call made inside a `ctx.step` or `@tool` body is metered but not journaled. The step or tool result is the durable unit, and its body runs again whole on a retry, so the call itself has no journal entry. - **Metered.** Before the call is sent it is held to the run's caps (`max_llm_calls`, `max_total_tokens`, `max_cost_usd`, and a case's cost cap); after it, one `llm:done` is written at once, so the spend survives a crash before the step ends. The event carries the enclosing unit's `step_id` and `call_index` and a `within` field: `"step"`, `"tool"`, or `"run"` for a `@tool` called straight from a workflow body, which has no step to charge. - **A cap behaves as it does outside a step.** A refusal raised in the body fails the step or tool like any other error: it is not retried, it is journaled as the unit's failure (a replay raises the same error and sends no request), and the run ends on `budget_exceeded` unless your code catches it. A body or a `try` around `ctx.step` that catches the refusal carries on, for example with a cheaper fallback; the cap stays reached, so its next model call is refused again. A case's cost cap (`live_budget_exceeded`) is routed the same way. - **Replay.** A step that finished replays its recorded result, so its calls are not made again and its `llm:done` events are not repeated. The spend and the `max_llm_calls` count are rebuilt from the journal. A step that was open at a crash runs again, and its new calls are metered again, because they are billed again. - **Streams.** A streamed call is held to the caps before it opens and metered once: an exact `llm:done` when it is read to its end, an `estimated` one when it is closed, dropped or fails first, as for a journaled stream. It is not recorded, so a replay never reads it again. - **Late replies.** A sync tool that times out leaves its request running on its thread. If the reply arrives while the run is live it is metered once; after the run's terminal event nothing is written, so that spend is not counted and a WARNING (`model_spend_after_terminal`, with the model and token counts) says so. - **Server-side tool fees.** Per-request fees for server-side tools (Anthropic MCP and server tools, OpenAI hosted tools, Bedrock server tools) are not metered; only token usage is. Gemini automatic function calling stays refused inside a step or tool, since it would hide several model requests behind one cap check. - **Not covered.** The SDK surfaces that are forwarded unwrapped (`completions`, `audio`, `with_raw_response`, ...) pass through unmetered inside a step or tool. A call inside a Pydantic AI model request is already counted by the adapter and is not counted twice. ### Sync code and threads `pyrula.agents.compat.openai.OpenAI` (also `from pyrula.agents.compat import OpenAI`) is the sync twin: a subclass of `openai.OpenAI` with the same journaled, passed-through and refused surfaces as `AsyncOpenAI`, including `rate_limit=`, `copy()` and `with_options()`. A built `openai.OpenAI` converts in place with `journaled(client)`. The sync Anthropic client, `pyrula.agents.compat.Anthropic`, follows every rule below the same way (`AsyncAnthropic` is its async twin, and `journaled(anthropic.Anthropic(...))` converts a built one). A run's body is async, and the run's event loop does all its journaling. A sync client therefore journals only when it is called on a thread of the run: `await ctx.to_thread(fn)` for one call and `await ctx.map_threads(fn, items)` for a fan-out. The thread carries the run's context, and each call gets its own replay scope (`thread:`, `map::`), so concurrent threads have stable keys and a resume replays each one. See [Blocking calls in threads](https://docs.pyrula.io/workflows/guides/parallel-and-children.md#blocking-calls-in-threads). ```python import asyncio import json import httpx from pyrula.agents import AgentContext, agent from pyrula.agents.compat.openai import OpenAI from pyrula.agents.testing.runner import AgentRunner provider_calls = [] def provider(request: httpx.Request) -> httpx.Response: # A stand-in for the OpenAI API, so this block runs offline. provider_calls.append(json.loads(request.content)) return httpx.Response(200, json={ "id": "chatcmpl-1", "object": "chat.completion", "created": 0, "model": "gpt-6-luna", "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "Pong."}}], "usage": {"prompt_tokens": 9, "completion_tokens": 2, "total_tokens": 11}, }) @agent(name="sync_ping") async def ping(ctx: AgentContext) -> str: client = OpenAI( api_key="unused", base_url="http://openai.test/v1", http_client=httpx.Client(transport=httpx.MockTransport(provider)), ) def ask() -> str: reply = client.chat.completions.create( model="gpt-6-luna", messages=[{"role": "user", "content": "Ping?"}] ) return reply.choices[0].message.content return await ctx.to_thread(ask) events = asyncio.run(AgentRunner(ping).run()) assert events[-1].payload["result"] == "Pong." and len(provider_calls) == 1 ``` - **On the loop thread.** A sync call made directly in the async body would block the run's heartbeats and cannot be journaled, so it raises `UnjournaledSurface` before any request is sent. The message names the fix: `await asyncio.to_thread()` (or `await ctx.to_thread()`), or the async twin (`AsyncOpenAI`, `AsyncAnthropic`). - **Outside a run,** a sync call is a passthrough, as for the async client. **Inside a `ctx.step` or `@tool` body** a call is metered, not journaled: see "Calls inside a step or tool" above. A sync call written directly in an `async` step or tool body runs on the loop thread and cannot be metered, so it raises `UnjournaledSurface` before any request is sent. Call it as `await asyncio.to_thread(fn)` (or `await ctx.to_thread(fn)`) there, or use the async client. - **A thread started without the run's context** (`threading.Thread`, `ThreadPoolExecutor`, `run_in_executor`) sees no run. While a run is bound in the process, the call logs a warning once per call site naming `ctx.to_thread` and `ctx.map_threads`, and it runs live again on every resume. - **Streams.** A sync stream supports `for`, `with` and `close()`, live and on replay, and is not the SDK's `Stream` (no `.response`). Journaling and abandonment follow the async rules: a stream you stop reading earlier is metered once, marked `estimated`, and runs live again on resume. - **Cancellation.** The request runs on the calling thread. If the run is cancelled or its attempt ends while the request is in flight, the thread's request finishes on its own, is billed and is not journaled. A sync call can bill a whole reply where the async client, which aborts the HTTP request, may bill part of it. The thread's next call into the run raises `RunAttemptEnded`. - **Rate limits.** `rate_limit=` reserves its slot through the run's loop, for a journaled call on a thread of the run and for a passthrough call on a thread inside a step or tool. The one case with no slot is a sync call with no run bound, which runs unlimited, on the SDK's own retries, and logs a warning. On the loop thread the call is refused, as above. A streamed call is charged its estimate; its usage is not reported back to the slot. #### instructor The journaled clients work unchanged under [instructor](https://python.useinstructor.com/): each attempt, a reask after a validation error included, is one journaled `create` whose messages (the reask text too) are in its fingerprint, so a resume replays every attempt with no second request. `TOOLS`, `JSON_SCHEMA` and `RESPONSES_TOOLS` modes (with a reask) are tested with the sync client in `ctx.to_thread` and with the async client; `create_partial` streaming is tested with the sync client in `TOOLS` and `RESPONSES_TOOLS`. Tested versions: instructor 1.17.0 on the locked openai (2.41) and on openai 3.3.0, the newest release that resolves with it. instructor 1.17 caps `jiter` below 0.15 and openai 3.4 and later need a newer one, so instructor on openai 3.4 or later is untested until instructor relaxes the cap. ```python import instructor from pyrula.agents.compat import OpenAI patched = instructor.from_openai(OpenAI()) # the drop-in, wrapped by instructor extracted = await ctx.to_thread( lambda: patched.chat.completions.create(model="gpt-6-luna", response_model=User, messages=[...]) ) ``` instructor captures the client's bound `chat.completions.create` when you wrap the client. Journal first: | Order | Result | |---|---| | `instructor.from_openai(compat.OpenAI())` | Journaled. | | `instructor.from_openai(journaled(openai.OpenAI()))` | Journaled. | | `instructor.from_openai(compat.AzureOpenAI(...))` | Journaled. | | `journaled(client)` after `instructor.from_openai(client)` | instructor keeps the SDK's unjournaled method: its calls run live again on every resume. | | `journaled(instructor.from_openai(client))` | `TypeError`: pass the provider client, before `from_openai`. | The module-level client (`openai.chat.completions.create`) is not covered; construct `compat.OpenAI()` instead. #### Azure OpenAI and Bedrock `AzureOpenAI` and `AsyncAzureOpenAI` (and, on an openai release that has them, `BedrockOpenAI` and `AsyncBedrockOpenAI`) come from `pyrula.agents.compat` the same way: each is a subclass of the SDK's class with the surfaces, `rate_limit=`, `copy()` and `with_options()` of `OpenAI` and `AsyncOpenAI`, and `journaled(client)` converts an Azure or Bedrock client you built. On an openai release without Bedrock, importing a Bedrock name raises `ImportError` naming the release. `OpenAI(provider=...)` (openai 3.x only) is the SDK's own constructor and is journaled like any `OpenAI`. To detect Bedrock support, test `hasattr(openai, "BedrockOpenAI")`, not the compat module. An Azure `model` is a deployment name, so its price is known only once a reply has named the model. The first call of a deployment under `max_cost_usd` skips the "no price" check; the reply's `model` (the underlying model, such as `gpt-4o`) prices it from then on, unless the deployment name has a price row of its own, and `llm:done` records the deployment as `requested_model`. A reply whose model has no price stops the next costed call with `unpriced_model`. A streamed Azure call always sends `stream_options.include_usage`, because the meter needs the usage chunk. Whether every Azure `api_version` accepts it is not verified yet (the live Azure check, B102, will); until then, pin an `api_version` that does. Binary inputs (images, PDFs) of an abandoned Azure stream are counted at the largest image bound of any OpenAI model, since a deployment name does not say which one it is. A Bedrock model id is canonicalised (`openai.gpt-oss-120b-1:0` is priced as `gpt-oss-120b`); a Bedrock application inference profile ARN is opaque like an Azure deployment. ### Anthropic `pyrula.agents.compat.anthropic.AsyncAnthropic` and the sync `Anthropic` (`pip install "pyrula[agents]"`) journal `messages.create` and `beta.messages.create`, streaming or not, and rebuild the `Message` (or `BetaMessage`) or the stream events on replay. A stream is journaled when `message_stop` arrives; truncation and failure follow the chat stream rules. `messages.batches.create`, `beta.messages.batches.create` and (anthropic 2.x) `beta.dreams.create` are journaled as writes: a replay returns the recorded object instead of creating another job. The sync client journals on a thread of the run, as [described above](#sync-code-and-threads) for OpenAI. - **The SDK's helpers.** `messages.stream`, `messages.parse`, `beta.messages.stream` and `beta.messages.parse` are journaled as their `create` plus the SDK's own builder (`parse` exists only on an anthropic release that has it; on 0.47 `hasattr(client.messages, "parse")` is `False`). The parsed message is rebuilt by the SDK's parser from the stored reply and the stream manager (`text_stream`, `get_final_message()`, events) runs over the journaled stream; neither makes a second provider request. `output_format=` is part of the replay check, a reply the parser rejects replays as the same error, your model's validators run again on replay, and `request_id` is `None` on a replayed message. A manager you never enter sends and journals nothing. A stream you stop early, cancel or drop is metered once as an estimate. On anthropic 0.7x only the beta managers take `output_format`, so there `output_format=` streams and `parse` use the SDK's own helper outside a run and are refused inside one; a plain `messages.stream()` is journaled on every release. - **Refused in a run:** `beta.messages.thread` (the helper, not the `thread=` argument below) and `beta.messages.tool_runner`, `beta.sessions`, `beta.deployments`, `with_raw_response`, `with_streaming_response` and the legacy `completions`. Their replay would call the provider again outside the journal. Config surfaces like `beta.agents` pass through. - **Server-side tools that act are refused,** on `messages.create` and `beta.messages.create` alike: `mcp_servers`, `mcp_toolset`, the `code_execution` tools (including `bash_code_execution` and `text_editor_code_execution`), `advisor_*`, `browser_toolset`, `container={"skills": ...}` and any tool type this list does not know (a type is matched by its exact family name plus an optional date, so a lookalike such as `memory_server_*` is refused). Anthropic's servers run them, so a crash before the journal write, a what-if branch or a re-run would run them again; a journal replays answers, not their side effects. The error names the argument. Put the call inside `ctx.step` or a `@tool` (the step replays its result), or leave the tool out. Custom tools, `web_search`, `web_fetch`, `tool_search` and the client-executed `bash`, `text_editor`, `computer` and `memory` tools are allowed: your code runs the last four, and the first three only read. This matches the OpenAI Responses rule. Before this release `messages.create` forwarded these tools; it now refuses them. - **Replay check.** A resumed call is compared on `system`, `messages`, `tools`, `tool_choice`, `output_config` and `container`; a beta call also on `betas`, `output_format`, `context_management`, `compaction`, `fallbacks` and the `thread=` argument. Model, sampling, `thinking` and `metadata` are not compared. - **Server-held state.** A call that continues a `thread=` or reuses a `container` depends on state Anthropic holds. The journal records the reply, not that state, so the first live call after the replayed prefix needs it to still exist. - **Middleware.** A call on a client with middleware (anthropic 2.x `middleware=` or `with_middleware`) is refused in a run, since a middleware can send a request more than once and one journaled call would hide the rest. None is classified as sending once yet. - **Metering.** Usage is read from `usage` (a stream: `message_start`, then the cumulative `message_delta`), cache reads and writes included. A stream stopped early is metered once, marked `estimated`, with images and PDFs in the request counted at Anthropic's per-image and per-page upper bounds. #### Anthropic on Bedrock and Vertex `AnthropicBedrock`, `AsyncAnthropicBedrock`, `AnthropicVertex` and `AsyncAnthropicVertex` come from `pyrula.agents.compat` the same way (`pip install "anthropic[bedrock]"` or `"anthropic[vertex]"`, which bring boto3 and google-auth). Each is a subclass of the SDK's own class, which is not a subclass of `Anthropic`, with the surfaces, refusals, replay check, `rate_limit=` and sync rules of the section above. `journaled(client)` converts a client you built and keeps the `messages` and `beta` resources the SDK assigned. Foundry and AWS clients are not converted yet. ```python from pyrula.agents.compat import AsyncAnthropicBedrock, AsyncAnthropicVertex bedrock = AsyncAnthropicBedrock(aws_region="us-east-1") vertex = AsyncAnthropicVertex(region="us-east5", project_id="my-project") reply = await bedrock.messages.create( model="anthropic.claude-haiku-4-5-20251001-v1:0", max_tokens=256, messages=messages ) ``` - **Prices.** A Bedrock or Vertex id is priced by the id you asked for, then its canonical form: `anthropic.claude-haiku-4-5-20251001-v1:0`, `us.anthropic.claude-haiku-4-5-20251001-v1:0` and `claude-haiku-4-5@20251001` are priced as `claude-haiku-4-5`, the Anthropic row, unless a row keyed on the cloud's own id exists (regional premiums are not in the packaged table). `llm:done` records the canonical id as `model` and your id as `requested_model`. An application inference profile ARN names no model: its first call skips the "no price" check and the reply's model prices the next. - **Bedrock stream metrics.** The last event of a Bedrock stream carries `amazon-bedrock-invocationMetrics`. Usage is read from `message_start` and `message_delta` alone, so those counts are not added; the key stays on the recorded event and comes back on replay. - **Optional packages.** Importing a name needs no boto3 or google-auth. A Bedrock client with an `api_key` (or `AWS_BEARER_TOKEN_BEDROCK`) never needs boto3; SigV4 signing does, and the first call without it raises `ImportError` naming `anthropic[bedrock]`. A Vertex client without google-auth fails in the SDK, which names `anthropic[vertex]`. - **Local failures are not journaled.** A failure raised before a request is sent (no AWS or Google credentials, a request that could not be signed, a missing boto3) says nothing about the model, so it is re-raised without a journal entry and a resume asks again. A status error, a timeout or a connection error is journaled and replays as the same class, as for `Anthropic` and `OpenAI`, a refused connection included: a resume replays that error instead of retrying, whereas `journaled(boto3_client).converse` leaves a connect failure unjournaled. A call that fails locally still takes its rate-limit estimate. - **Not supported by the cloud.** Bedrock has no batch API, so `messages.batches.create` raises the SDK's own error before any request. Server-side tools that act are refused as above. - **`rate_limit=`.** Requests that wait for a slot are sent by a copy of the client with retries off. The SDK's `copy()` of the Bedrock client leaves out `aws_profile` and builds a new HTTP client, so the drop-in rebuilds it with both, and a Bedrock client signs with its profile and sends through your `http_client`. `copy()` and `with_options()` on the drop-in keep the profile and the HTTP client for the same reason. #### instructor with Anthropic The same rules as for OpenAI: journal first, then `instructor.from_anthropic`. `TOOLS` and `JSON` modes (with a reask) are tested with the sync client in `ctx.to_thread` and with the async client, `create_partial` streaming in `TOOLS`, and `beta=True` in `TOOLS` mode, which sends `beta.messages.create`. instructor 1.17's `JSON` and `JSON_SCHEMA` modes reject the `BetaMessage` that `beta.messages.create` returns, with or without pyrula, so they do not combine with `beta=True`. Tested versions: instructor 1.17.0 on the locked anthropic, on anthropic 2.3.0 and on the 0.47.0 floor. instructor takes the Bedrock and Vertex drop-ins too (`instructor.from_anthropic(compat.AnthropicBedrock(...))`, a reask included); it does not accept Foundry or AWS clients. | Order | Result | |---|---| | `instructor.from_anthropic(compat.Anthropic())` | Journaled. | | `instructor.from_anthropic(journaled(anthropic.Anthropic()))` | Journaled. | | `journaled(client)` after `instructor.from_anthropic(client)` | instructor keeps the SDK's unjournaled method: its calls run live again on every resume. | | `journaled(instructor.from_anthropic(client))` | `TypeError`: pass the provider client, before `from_anthropic`. | ### Gemini `pyrula.agents.compat.genai` (`pip install "pyrula[gemini]"`) replaces `from google import genai`: ```python from pyrula.agents.compat import genai client = genai.Client(api_key="...") ``` `genai.Client` is a subclass of `google.genai.Client`, so `isinstance` checks and every constructor argument are the SDK's; `types`, `errors` and every other name forward to `google.genai`. `models.generate_content` and `models.generate_content_stream` are journaled on the sync client and on `client.aio`, and so is every `chats` turn, since the SDK builds a chat over `models`. `batches.create`, `batches.create_embeddings` and `tunings.tune` are journaled as writes: a replay returns the recorded job instead of creating another. A client you built converts in place with `journaled(client)`; read `client.aio` after that, since an `aio` reference taken before it stays the SDK's own and unjournaled. The sync client journals on a thread of the run, as [described above](#sync-code-and-threads) for OpenAI, and raises `UnjournaledSurface` on the run's own loop thread. The SDK's client has no `copy` or `with_options`, and neither has this one. - **Refused in a run:** `models.generate_images`, `edit_image`, `upscale_image`, `generate_videos`, `recontext_image` and `segment_image`, `aio.live`, and the `interactions`, `agents` and `triggers` APIs. A replay of any of them would call the provider again outside the journal. Embeddings, token counts, model, file, cache, tuning reads and cancel, and store calls pass through: they are not generation. - **Tools.** A Python function or an MCP session in `config.tools` is refused unless `automatic_function_calling={"disable": True}` is set (or `maximum_remote_calls` is 0 or less): with it on, the SDK runs your function and calls the model again inside one `generate_content`, which a replay cannot skip. Pass function declarations and run the calls yourself in a `ctx.step` or `@tool`, or use an `@agent` with Gemini tools. The tools that act on a server are refused too: `code_execution`, `mcp_servers` and any `Tool` field this list does not know. Allowed: `function_declarations`, `computer_use` (your code performs the actions), and the tools that only read: `google_search`, `google_search_retrieval`, `google_maps`, `url_context`, `file_search`, `retrieval`, `enterprise_web_search`, `exa_ai_search` and `parallel_ai_search`. A Python function with automatic calling off is fingerprinted by its name, signature and docstring, so changing a parameter refuses a resume; an MCP session has no stable form and is not fingerprinted. A replay runs none of them again, but a crash before the journal write or a what-if branch asks Google to search again, and the results can differ. - **Replay check.** A resumed call is compared on `contents` and, from `config`, `system_instruction`, `tools`, `tool_config`, `response_schema`, `response_json_schema`, `response_mime_type` and `cached_content`. Inline bytes (a PDF, an image, audio) and a PIL image are compared by a digest, never stored. Model, sampling, `thinking_config`, `safety_settings`, `candidate_count` and `seed` are not compared. - **The replayed reply** is the journaled `GenerateContentResponse` (`parsed` rebuilt when `response_schema` is a type), without `sdk_http_response` and `automatic_function_calling_history`. A journal written by a newer google-genai is read by an older one with the fields it does not declare dropped. - **Automatic continuation** (google-genai 2.x, on by default) is the SDK sending follow-up requests inside one call. The merged reply is one journaled call with the summed usage; a replay sends none of them. - **Rate limit.** `genai.Client(rate_limit=LLMRate(...))` takes a slot per live call, sync and `aio`, under the same default key as `GeminiLLM`, so a drop-in and a native client share a queue. The SDK has no `max_retries` and does not retry unless you set `HttpOptions(retry_options=...)`, per client or per request; those retries and continuation hops happen inside the one slot. - **Server-held state.** `cached_content` names and uploaded files live at Google. The journal records the reply, not that state, so the first live call after the replayed prefix needs it to still exist. - **Metering.** Usage is read from `usage_metadata` (a stream: the last chunk that has it, which is cumulative): prompt and tool-use prompt tokens less cached tokens are input, cached tokens are cache reads, candidates and thoughts tokens are output. A stream stopped early is metered once, marked `estimated`, with images and PDFs in the request counted at Gemini's per-image and per-page upper bounds and function-call arguments counted as output. #### instructor with Gemini Journal first, then `instructor.from_genai`: it calls `client.models` or `client.aio.models` when you call it, so it composes with `compat.genai.Client()` and with `journaled(client)` called before it. Tested with instructor 1.17.0 in `GENAI_TOOLS` mode, the sync client in `ctx.to_thread` and the async client, on the locked google-genai and on the 1.46.0 floor. Calling `journaled(client)` after `from_genai` is not tested. ### Amazon Bedrock (boto3) `journaled(client)` takes the boto3 `bedrock-runtime` client you already built (`pip install "pyrula[bedrock]"`) and makes `converse` and `converse_stream` journaled, in place: ```python import boto3 from pyrula.agents.compat import journaled client = journaled(boto3.client("bedrock-runtime", region_name="us-east-1")) reply = await ctx.to_thread(lambda: client.converse(modelId=model_id, messages=messages)) ``` The client is still a boto3 client, with your profile, region and `Config`. boto3 is blocking, so a call journals on a thread of the run ([described above](#sync-code-and-threads)) and raises `UnjournaledSurface` on the run's own loop thread; aioboto3 is not supported. A resumed run replays the recorded reply (a stream's events, `ResponseMetadata` stubbed) and makes no second request. `rate_limit=LLMRate(...)` takes a slot per live call under the same default key as `BedrockLLM`, so the two share a queue; botocore's own retries happen inside that one slot and one journaled call. Call `journaled(client)` before `instructor.from_bedrock` or anything else that captures `client.converse`, and not on a method you already took a reference to. - **Refused in a run:** `invoke_model`, `invoke_model_with_response_stream`, `invoke_model_with_bidirectional_stream` and `start_async_invoke`, and any operation a newer boto3 adds that is not classified: a replay would call the provider again outside the journal. Use `converse`. `count_tokens`, `apply_guardrail`, `get_async_invoke` and `list_async_invokes` pass through and are not journaled, so a resume sends them again. `apply_guardrail` can be model-backed (guardrail filters and contextual grounding are billed), and a guardrail edited between attempts can answer differently: run it in a `ctx.step` when its verdict decides what the run does. - **Tools.** Your own `toolSpec` tools are allowed: the model asks, your code runs the call in a `ctx.step` or `@tool`. A server-run `systemTool` is refused except the read-only `nova_grounding`. - **Replay check.** A resumed call is compared on `messages`, `system`, `toolConfig`, `outputConfig`, `promptVariables` and `guardrailConfig`. Image, document, video and audio bytes are compared by a digest, never stored. `modelId`, `inferenceConfig`, `additionalModelRequestFields`, `additionalModelResponseFieldPaths`, `requestMetadata`, `serviceTier` and `performanceConfig` are not compared. - **Errors.** A service error (throttling, validation, access denied) and an error event in a stream are journaled once and replay as a `ClientError` with the recorded `Error` code, status and request id; `except client.exceptions.ThrottlingException` still matches. The rest of the response, headers included, is not kept. Failures botocore raises before the request goes out are not journaled, so a resume runs the call live: a request botocore rejects (`ParamValidationError`; `Config(parameter_validation=False)` turns that check off here too), missing credentials or region, and a connection that could not be made (`EndpointConnectionError`, `ConnectTimeoutError`). A read timeout is journaled, because the request was sent and may be billed; it replays as `ReadTimeoutError`, as does a transport failure in the middle of a stream (`ResponseStreamingError`), so a resume does not send the call twice. - **Metering.** Usage comes from the reply's `usage` (a stream: the `metadata` event), with cache read and write tokens counted as cache tokens. Spend is priced by the canonical id of `modelId`; an application inference profile ARN names no model, so it is priced only if you add a price for it. A stream stopped early is metered once, marked `estimated`, with images and documents counted at the Claude family's per-image and per-page bounds (the largest bound of any family for other models) and tool-input deltas counted as output. #### instructor with Bedrock Journal first, then `instructor.from_bedrock(client, mode=instructor.Mode.BEDROCK_TOOLS)`: it reads `client.converse` when you call it. Tested with instructor 1.17.0, the sync client in `ctx.to_thread`, on the locked boto3 and on the 1.38.0 floor. The async form (`async_client=True`) calls boto3 on the event loop, which a run refuses. The drop-ins and `durable_chat_model` count toward `max_llm_calls`, `max_total_tokens` and `max_cost_usd` ([Spend caps](https://docs.pyrula.io/agents/authoring/agents.md#spend-caps)). A side effect your code runs after the model replies (a tool call inside a LangGraph node) reruns on resume; make it a `@tool` or `ctx.step`. ## build_mcp_server Exposes Pyrula tools over MCP, running each call as a durable workflow. Host setup for Claude Code and Cursor is in [Serving tools over MCP](https://docs.pyrula.io/agents/integrations/mcp-server.md). Ships with `pip install "pyrula[agents]"`. ```python from pyrula.agents.mcp import build_mcp_server server = build_mcp_server([charge_card, send_email], store=store) server.run_stdio() # or server.run_http(auth=...) ``` The server submits the run and waits; a separate worker on the same store executes it, so a crash mid-tool resumes when a worker restarts. - Tools must be context-generic (no `ctx.llm`, no sub-agents). A tool whose first parameter is annotated `ToolContext` gets one with no turn bound (`ctx.idempotency_key` is `None`); others get a `WorkflowContext`. - `read_only=True` tools are advertised with MCP `readOnlyHint`. - A `requires_approval=True` tool raises `ToolRegistrationError`; a predicate-gated tool is built and fails per call when its condition holds. - The tool call is one durable step: the body cannot call `ctx.step`, `ctx.interrupt`, timers or signals. Idempotency depends on the client. With a stable `_meta: {idempotency_key: "..."}`, a resubmission after a crash re-attaches the existing run; the token is scoped to the caller (the authenticated principal, or what `identity_from_request` returns) and to the tool, and reusing it with different arguments fails with `IdempotencyConflict`. Without one, every call is its own run and a retry runs the tool again; the JSON-RPC request id is not a fallback, since ids restart per session and two clients would read each other's result. Pass `idempotency_from_request` to derive a token from any request field.