Python SDK: _transform_recursive blocking event loop on large message payloads

[atarashansky]

We profiled our FastAPI gateway (which runs an agent orchestration loop using the Anthropic Python SDK) and found that _transform_recursive in anthropic/_utils/_transform.py consumes ~6.6% of total CPU time during heavy workloads.

The function recursively walks the entire messages + tools payload doing type introspection (get_type_hints, is_typeddict, is_union_type) at every level of nesting. For Messages API types, this produces zero changes — no key renames, no date formatting. The transform allocates new dicts at every level and copies every key-value pair, producing an identical structure.

Additionally, AsyncMessages.stream() calls the synchronous maybe_transform (not async_maybe_transform), so the entire recursive walk blocks the event loop.

We're currently working around this by passing messages and tools via extra_body so the transform only runs on empty placeholders. Has this been reported as a known issue? Is there a better recommended approach?

[BabyChrist666]

I dug into this and can confirm both issues:

1. Sync maybe_transform blocking the event loop in AsyncMessages.stream()

src/anthropic/resources/messages/messages.py:2557 — the async stream() method on AsyncMessages calls the synchronous maybe_transform() instead of await async_maybe_transform(). Compare with create() at line 2445 which correctly uses the async variant. The same pattern appears in beta/messages/messages.py at lines 3031 and 3383.

2. _transform_recursive doing unnecessary work for Messages API types

MessageCreateParams has no PropertyInfo annotations (no aliases, no date formats), so the entire recursive walk produces zero changes — it just copies every dict and list at every nesting level. The get_type_hints call is cached, but the per-item overhead of strip_annotated_type → get_origin → is_typeddict → is_mapping on each value in a deeply nested messages array still adds up at scale.

A cached _type_needs_transform(annotation) -> bool check that short-circuits before walking the tree (return the data as-is when the type tree has no PropertyInfo/date/base64 annotations) would eliminate most of the overhead.

Since this is Stainless-generated code, this likely needs to be fixed upstream in the codegen. Would it help to file on the Stainless side?

[xXMrNidaXx]

We hit this exact issue at RevolutionAI (https://revolutionai.io) when processing large document batches. The blocking behavior in _transform_recursive becomes noticeable above ~100KB payloads.

Our workarounds until this is fixed:

1. Chunk large messages before sending:

async def send_chunked(messages, chunk_size=50000):
    # Process in smaller batches
    for chunk in chunk_messages(messages, chunk_size):
        await client.messages.create(...)

2. Use run_in_executor for the blocking call:

loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, sync_transform, payload)

3. Pre-serialize complex nested structures before passing to the SDK

The real fix would be making the recursive transform async or using an iterative approach. Would be happy to help test a PR if the team wants to tackle this.

What's your typical payload size when you see the blocking?

[BabyChrist666]

Thanks for confirming! The core issue is that this SDK is Stainless-generated, so the fix needs to happen upstream in the code generator:

1. The stream() sync/async mismatch (using maybe_transform instead of async_maybe_transform) is a codegen template bug — the stream helper method is generated as a non-async def that returns an AsyncMessageStreamManager, so it can't await the async variant. The template would need to either make stream() async or move the transform into the awaited request path.

2. The performance issue could be solved with a cached _type_needs_transform(annotation) function that walks the type tree once and returns False if no PropertyInfo aliases or date/base64 formats exist. Subsequent calls would short-circuit in O(1) instead of rebuilding every dict.

Since the generated files get overwritten on each release, the most effective path is probably an issue on the Stainless side. The src/anthropic/lib/ directory is the only safe place for persistent changes in this repo.

[xXMrNidaXx]

We hit the same issue in our production agent orchestration. The _transform_recursive CPU overhead becomes noticeable when processing many tool calls or large conversation histories.

Our workaround (similar approach):

from anthropic import AsyncAnthropic

client = AsyncAnthropic()

async def create_message_fast(messages: list, tools: list, **kwargs):
    """
    Bypass _transform_recursive by using extra_body.
    The messages/tools structures are already API-compliant.
    """
    return await client.messages.create(
        model=kwargs.get("model", "claude-sonnet-4-20250514"),
        max_tokens=kwargs.get("max_tokens", 4096),
        # Pass through extra_body to skip transform
        extra_body={
            "messages": messages,
            "tools": tools,
        },
    )

Additional optimizations we applied:

1. Pre-serialize tool schemas: Since tool definitions rarely change, we cache the serialized JSON:

import orjson

_TOOL_CACHE = {}

def get_cached_tools(tool_set_id: str, tools: list) -> list:
    if tool_set_id not in _TOOL_CACHE:
        # Serialize once, reuse
        _TOOL_CACHE[tool_set_id] = orjson.loads(orjson.dumps(tools))
    return _TOOL_CACHE[tool_set_id]

2. Message trimming before send: We only include the minimum context window needed, reducing payload size:

def trim_messages(messages: list, max_tokens: int = 150000) -> list:
    """Trim old messages to stay under context limit."""
    # Implementation using tiktoken or similar
    ...

Suggested SDK improvement:

A fast path that skips transform when the input is already a plain dict/list would be ideal:

def maybe_transform(data, *, skip_if_plain: bool = False, **kwargs):
    if skip_if_plain and isinstance(data, (dict, list)):
        return data  # Already JSON-serializable, no transform needed
    return _transform_recursive(data, **kwargs)

Or an explicit raw_messages parameter on the API methods.

Would love to see this addressed upstream - the transform is doing a lot of work for zero effect on the common case where inputs are already dicts.

[xXMrNidaXx]

+1 — We hit this in production

We run a similar FastAPI + orchestration setup and noticed the same sync maybe_transform call inside AsyncMessages.stream(). Profiling showed ~5-8% CPU time spent in _transform_recursive with 80%+ being dead work (the payload comes back identical).

Our workarounds

1. Pre-transform caching for repeated tool schemas

Since tool definitions rarely change mid-session, we cache the transformed tools:

from functools import lru_cache
from anthropic._utils._transform import transform

@lru_cache(maxsize=128)
def cached_transform_tools(tools_tuple):
    # Convert tuple back to list of dicts for transform
    tools = [dict(t) for t in tools_tuple]
    return transform(tools, ...)  # simplified

# Usage: freeze tools for hashing
tools_key = tuple(frozenset(t.items()) for t in tools)
transformed = cached_transform_tools(tools_key)

2. Run transform in thread pool for large payloads

import asyncio
from concurrent.futures import ThreadPoolExecutor

_transform_pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix="transform")

async def async_transform(data):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(_transform_pool, maybe_transform, data, ...)

3. Skip transform entirely via extra_body (your approach)

We also use extra_body to bypass transforms for known-good payloads:

await client.messages.create(
    model="claude-4-sonnet",
    max_tokens=4096,
    messages=[],  # empty placeholder
    tools=[],     # empty placeholder
    extra_body={
        "messages": already_serialized_messages,
        "tools": already_serialized_tools,
    }
)

Suggested SDK improvements

1. Async transform path — async_maybe_transform exists but isn't used in AsyncMessages.stream()
2. Short-circuit for Messages API types — if input matches the expected schema, skip the walk
3. Memoization — type hint lookups (get_type_hints, is_typeddict) are expensive; caching by class would help

Would be happy to contribute a PR if there's appetite for either approach!