Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import asyncio
import logging
import sys
from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence
from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence, Set
from dataclasses import dataclass, field
from typing import Any, ClassVar, Final, Generic, Literal, TypedDict, cast

Expand Down Expand Up @@ -284,6 +284,7 @@ class _AnthropicRequestState:

active_call_id: str | None = None
active_call_content_type: str | None = None
active_call_is_hosted_function: bool = False
tool_name_aliases: dict[str, str] = field(default_factory=dict[str, str])


Expand Down Expand Up @@ -901,9 +902,11 @@ def _prepare_messages_for_anthropic(self, messages: Sequence[Message]) -> list[d
else:
msgs = list(messages)

pending_hosted_calls = self._pending_hosted_function_calls(msgs)

result: list[dict[str, Any]] = []
for msg in msgs:
result.extend(self._prepare_message_groups_for_anthropic(msg))
result.extend(self._prepare_message_groups_for_anthropic(msg, pending_hosted_calls))

# Anthropic requires the conversation to end with a user message.
# Append a synthetic user turn so chained agent outputs work as
Expand All @@ -913,9 +916,43 @@ def _prepare_messages_for_anthropic(self, messages: Sequence[Message]) -> list[d

return result

def _prepare_message_groups_for_anthropic(self, message: Message) -> list[dict[str, Any]]:
@staticmethod
def _pending_hosted_function_calls(messages: Sequence[Message]) -> set[int]:
"""Return the identities (``id()``) of hosted function calls that have no result yet.

A hosted (server tool) call without a result is still running on Anthropic's side, for example
when Claude called a local tool in parallel. It has to be sent back as ``server_tool_use``, or
Anthropic expects a local ``tool_result`` for it.

A ``call_id`` is not unique across a transcript: a later occurrence may reuse it. Calls and results
are therefore correlated in transcript order. A result settles the open call occurrence with the
same ``call_id`` whose ``Content.id`` it carries, and otherwise the earliest open occurrence with
that ``call_id``. A result never settles a call that appears after it.
"""
open_calls_by_call_id: dict[str, list[Content]] = {}
for message in messages:
for content in message.contents:
if not content.call_id:
continue
if content.type == "function_call":
open_calls_by_call_id.setdefault(content.call_id, []).append(content)
elif content.type == "function_result" and (open_calls := open_calls_by_call_id.get(content.call_id)):
settled_index = next(
(
index
for index, call in enumerate(open_calls)
if content.id is not None and call.id == content.id
),
0,
)
open_calls.pop(settled_index)
return {id(call) for open_calls in open_calls_by_call_id.values() for call in open_calls if call.hosted}

def _prepare_message_groups_for_anthropic(
self, message: Message, pending_hosted_calls: Set[int] | None = None
) -> list[dict[str, Any]]:
"""Prepare a Message and split Anthropic content blocks into valid role groups."""
prepared_message = self._prepare_message_for_anthropic(message)
prepared_message = self._prepare_message_for_anthropic(message, pending_hosted_calls)
content = prepared_message.get("content")
if not isinstance(content, list):
return [prepared_message]
Expand Down Expand Up @@ -964,11 +1001,17 @@ def _message_has_tool_use(self, message: dict[str, Any]) -> bool:

return False

def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]:
def _prepare_message_for_anthropic(
self, message: Message, pending_hosted_calls: Set[int] | None = None
) -> dict[str, Any]:
"""Prepare a Message for the Anthropic client.

Args:
message: The Message to convert.
pending_hosted_calls: Identities (``id()``) of the hosted (server tool) call contents that have
no result yet, from ``_pending_hosted_function_calls``. These are sent as ``server_tool_use``
blocks. An identity rather than a ``call_id`` is used because a later occurrence may reuse
the ``call_id`` of a call that already has its result.

Returns:
A dictionary representing the message in Anthropic format.
Expand Down Expand Up @@ -1001,8 +1044,9 @@ def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]:
else:
logger.debug(f"Ignoring unsupported data content media type: {content.media_type} for now")
case "function_call":
pending_hosted_call = content.hosted and id(content) in (pending_hosted_calls or ())
a_content.append({
"type": "tool_use",
"type": "server_tool_use" if pending_hosted_call else "tool_use",

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shivam Sharma (@ShivamSharma43), the stream=True scenario from #9478 still loses the pending search's arguments. A streamed server_tool_use starts with input={}, and its query arrives through input_json_delta, but _parse_contents_from_anthropic discards those deltas for server tools. This branch consequently replays server_tool_use with input={} after the local result, rather than the original query, leaving the deferred search without its input. Please retain and accumulate the hosted argument deltas while keeping the call informational-only, and add a streaming regression that checks the query in the next outgoing request. The new tests supply complete inputs up front and therefore miss this path.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, you're right: the streamed path dropped the hosted argument deltas. Fixed in 5a09876.

  • _parse_contents_from_anthropic now keeps input_json_delta chunks for a server_tool_use call that is parsed as a hosted function call. They are emitted as hosted=True, informational_only=True fragments with the call's call_id and an empty name, so the existing streaming merge accumulates them into the call (the same mechanism as client-side tool_use deltas). The call stays informational-only and is not invoked locally. Deltas for MCP and code execution calls are still skipped, since those map to their own content types.
  • New streaming regression: test_streaming_parallel_server_tool_use_replays_streamed_input. It streams a local tool_use plus a server_tool_use that starts with input={} and gets its query through two input_json_delta chunks, runs the local tool, and asserts the second outgoing request replays server_tool_use with {"query": "agent framework release"} next to the local tool_result. It fails on the previous head with {} == {'query': ...}.
  • test_parse_contents_server_tool_use_input_json_delta_ignored asserted the old dropping behavior, so it is replaced by test_parse_contents_server_tool_use_input_json_delta_is_hosted_and_informational, plus test_parse_contents_mcp_and_code_execution_input_json_delta_ignored for the cases that are still skipped. test_concurrent_streams_isolate_tool_state_when_peer_fails now expects the hosted stream's delta fragment, and also checks that the concurrent local stream's fragments are not marked hosted.

"id": content.call_id,
"name": content.name,
"input": content.parse_arguments(),
Expand Down Expand Up @@ -1411,6 +1455,7 @@ def _parse_contents_from_anthropic(
case "tool_use" | "mcp_tool_use" | "server_tool_use":
request_state.active_call_id = content_block.id
request_state.active_call_content_type = content_block.type
request_state.active_call_is_hosted_function = False
if content_block.type == "mcp_tool_use":
contents.append(
Content.from_mcp_server_tool_call(
Expand All @@ -1437,6 +1482,7 @@ def _parse_contents_from_anthropic(
)
else:
resolved_tool_name = request_state.tool_name_aliases.get(content_block.name, content_block.name)
request_state.active_call_is_hosted_function = content_block.type == "server_tool_use"
contents.append(
Content.from_function_call(
call_id=content_block.id,
Expand Down Expand Up @@ -1682,16 +1728,20 @@ def _parse_contents_from_anthropic(
)
)
case "input_json_delta":
# Skip argument deltas for MCP and server tools — execution is handled server-side.
if request_state.active_call_content_type in ("mcp_tool_use", "server_tool_use"):
pass
else:
# MCP and code execution server tool calls are parsed into their own content types, so
# their argument deltas are skipped. Argument deltas of other server tool calls (such as
# web search) are kept as hosted, informational-only fragments that merge into the call,
# so a pending call is replayed with its full input.
hosted_call = request_state.active_call_is_hosted_function
if hosted_call or request_state.active_call_content_type not in ("mcp_tool_use", "server_tool_use"):
call_id = request_state.active_call_id or ""
contents.append(
Content.from_function_call(
call_id=call_id,
name="",
arguments=content_block.partial_json,
informational_only=hosted_call,
hosted=hosted_call,
merge_mode="append" if streaming else None,
raw_representation=content_block,
)
Expand Down
Loading
Loading