Skip to content

Python: Isolate Anthropic request parsing state #8237

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Eduard van Valkenburg (eavanvalkenburg) merged 3 commits into from
Sep 11, 2026
+383 −82
Merged

Python: Isolate Anthropic request parsing state #8237

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
905a605
Python: Isolate Anthropic request parsing state
eavanvalkenburg Sep 10, 2026
66455f5
Python: Cover concurrent Anthropic responses
eavanvalkenburg Sep 10, 2026
f870824
Merge branch 'main' into sturdy-pancake
eavanvalkenburg Sep 11, 2026
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
81 changes: 54 additions & 27 deletions python/packages/anthropic/agent_framework_anthropic/_chat_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import logging
import sys
from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass, field
from typing import Any, ClassVar, Final, Generic, Literal, TypedDict, cast

from agent_framework import (
Expand Down Expand Up @@ -255,6 +256,15 @@ def _map_finish_reason(stop_reason: str | None) -> FinishReason | None:
return FinishReason(FINISH_REASON_MAP.get(stop_reason, stop_reason))


@dataclass
class _AnthropicRequestState:
"""Mutable parsing state scoped to one Anthropic request."""

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


class AnthropicSettings(TypedDict, total=False):
"""Anthropic Project settings.

Expand Down Expand Up @@ -408,10 +418,6 @@ class MyOptions(AnthropicChatOptions, total=False):
self.anthropic_client = anthropic_client
self.additional_beta_flags = additional_beta_flags or []
self.model = model_setting
# streaming requires tracking the last function call ID, name, and content type
self._last_call_id_name: tuple[str, str] | None = None
self._last_call_content_type: str | None = None
self._tool_name_aliases: dict[str, str] = {}

# region Static factory methods for hosted tools

Expand Down Expand Up @@ -583,7 +589,8 @@ def _inner_get_response(
**kwargs: Any,
) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
# prepare
run_options = self._prepare_options(messages, options, **kwargs)
request_state = _AnthropicRequestState()
run_options = self._prepare_options(messages, options, request_state, **kwargs)

if stream:
# Streaming mode
Expand All @@ -595,7 +602,7 @@ async def _stream() -> AsyncIterable[ChatResponseUpdate]:
mark_feature_used(FeatureIndex.ANTHROPIC)
try:
async for chunk in await self.anthropic_client.beta.messages.create(**run_options, stream=True):
parsed_chunk = self._process_stream_event(chunk, emitted_usage)
parsed_chunk = self._process_stream_event(chunk, emitted_usage, request_state=request_state)
if parsed_chunk:
yield parsed_chunk
except AgentFrameworkException:
Expand All @@ -614,7 +621,7 @@ async def _get_response() -> ChatResponse:
raise
except Exception as ex:
raise _wrap_anthropic_error(ex) from ex
return self._process_message(message, options)
return self._process_message(message, options, request_state=request_state)
Comment thread
eavanvalkenburg marked this conversation as resolved.

return _get_response()

Expand All @@ -624,18 +631,22 @@ def _prepare_options(
self,
messages: Sequence[Message],
options: Mapping[str, Any],
request_state: _AnthropicRequestState | None = None,
/,
**kwargs: Any,
) -> dict[str, Any]:
"""Create run options for the Anthropic client based on messages and options.

Args:
messages: The list of chat messages.
options: The options dict.
request_state: State used to parse the response for this request.
kwargs: Additional keyword arguments.

Returns:
A dictionary of run options for the Anthropic client.
"""
request_state = request_state or _AnthropicRequestState()
# Start with a copy of options, excluding keys we handle separately
run_options: dict[str, Any] = {
k: v
Expand Down Expand Up @@ -697,7 +708,7 @@ def _prepare_options(
run_options["metadata"] = metadata

# tools, mcp servers and tool choice
if tools_config := self._prepare_tools_for_anthropic(options):
if tools_config := self._prepare_tools_for_anthropic(options, request_state=request_state):
run_options.update(tools_config)

# response_format - emit Anthropic's GA ``output_config.format`` shape.
Expand Down Expand Up @@ -1014,20 +1025,26 @@ def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]:
"content": a_content,
}

def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str, Any] | None:
def _prepare_tools_for_anthropic(
self,
options: Mapping[str, Any],
request_state: _AnthropicRequestState | None = None,
) -> dict[str, Any] | None:
"""Prepare tools and tool choice configuration for the Anthropic API request.

Converts FunctionTool to Anthropic format. MCP tools are routed to separate
mcp_servers parameter. All other tools pass through unchanged.

Args:
options: The options dict containing tools and tool choice settings.
request_state: State used to parse the response for this request.

Returns:
A dictionary with tools, mcp_servers, and tool_choice configuration, or None if empty.
"""
from agent_framework._types import validate_tool_mode

request_state = request_state or _AnthropicRequestState()
result: dict[str, Any] = {}
tools = options.get("tools")

Expand Down Expand Up @@ -1076,9 +1093,9 @@ def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str,
result["tools"] = tool_list
if mcp_server_list:
result["mcp_servers"] = mcp_server_list
self._tool_name_aliases = tool_name_aliases
request_state.tool_name_aliases = tool_name_aliases
else:
self._tool_name_aliases = {}
request_state.tool_name_aliases = {}

# Process tool choice
if options.get("tool_choice") is None:
Expand All @@ -1101,7 +1118,7 @@ def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str,
api_tool_name = next(
(
api_name
for api_name, local_name in self._tool_name_aliases.items()
for api_name, local_name in request_state.tool_name_aliases.items()
if local_name == required_name
),
required_name,
Expand All @@ -1124,12 +1141,18 @@ def _prepare_tools_for_anthropic(self, options: Mapping[str, Any]) -> dict[str,

# region Response Processing Methods

def _process_message(self, message: BetaMessage, options: Mapping[str, Any]) -> ChatResponse:
def _process_message(
self,
message: BetaMessage,
options: Mapping[str, Any],
request_state: _AnthropicRequestState | None = None,
) -> ChatResponse:
"""Process the response from the Anthropic client.

Args:
message: The message returned by the Anthropic client.
options: The options dict used for the request.
request_state: State used to parse this response.

Returns:
A ChatResponse object containing the processed response.
Expand All @@ -1139,7 +1162,7 @@ def _process_message(self, message: BetaMessage, options: Mapping[str, Any]) ->
messages=[
Message(
role="assistant",
contents=self._parse_contents_from_anthropic(message.content),
contents=self._parse_contents_from_anthropic(message.content, request_state=request_state),
raw_representation=message,
)
],
Expand All @@ -1151,7 +1174,10 @@ def _process_message(self, message: BetaMessage, options: Mapping[str, Any]) ->
)

def _process_stream_event(
self, event: BetaRawMessageStreamEvent, emitted_usage: dict[str, int] | None = None
self,
event: BetaRawMessageStreamEvent,
emitted_usage: dict[str, int] | None = None,
request_state: _AnthropicRequestState | None = None,
) -> ChatResponseUpdate | None:
"""Process a streaming event from the Anthropic client.

Expand All @@ -1161,6 +1187,7 @@ def _process_stream_event(
emitted, used to convert Anthropic's cumulative usage snapshots into
increments (see ``_incremental_usage``). Pass ``None`` for a one-off
event to keep the snapshot unchanged.
request_state: State used to parse this stream.

Returns:
A ChatResponseUpdate object containing the processed update.
Expand All @@ -1177,7 +1204,7 @@ def _process_stream_event(
role="assistant",
response_id=event.message.id,
contents=[
*self._parse_contents_from_anthropic(event.message.content),
*self._parse_contents_from_anthropic(event.message.content, request_state=request_state),
*usage_details,
],
model=event.message.model,
Expand All @@ -1201,13 +1228,13 @@ def _process_stream_event(
case "message_stop":
logger.debug("Received message_stop event; no content to process.")
case "content_block_start":
contents = self._parse_contents_from_anthropic([event.content_block])
contents = self._parse_contents_from_anthropic([event.content_block], request_state=request_state)
return ChatResponseUpdate(
contents=contents,
raw_representation=event,
)
case "content_block_delta":
contents = self._parse_contents_from_anthropic([event.delta])
contents = self._parse_contents_from_anthropic([event.delta], request_state=request_state)
return ChatResponseUpdate(
contents=contents,
raw_representation=event,
Expand Down Expand Up @@ -1270,8 +1297,10 @@ def _incremental_usage(cumulative: UsageDetails, emitted: dict[str, int] | None)
def _parse_contents_from_anthropic(
self,
content: Sequence[BetaContentBlock | BetaRawContentBlockDelta | BetaTextBlock],
request_state: _AnthropicRequestState | None = None,
) -> list[Content]:
"""Parse contents from the Anthropic message."""
request_state = request_state or _AnthropicRequestState()
contents: list[Content] = []
for content_block in content:
match content_block.type:
Expand All @@ -1284,8 +1313,8 @@ def _parse_contents_from_anthropic(
)
)
case "tool_use" | "mcp_tool_use" | "server_tool_use":
self._last_call_id_name = (content_block.id, content_block.name)
self._last_call_content_type = content_block.type
request_state.active_call_id = content_block.id
request_state.active_call_content_type = content_block.type
if content_block.type == "mcp_tool_use":
contents.append(
Content.from_mcp_server_tool_call(
Expand All @@ -1310,7 +1339,7 @@ def _parse_contents_from_anthropic(
)
)
else:
resolved_tool_name = self._tool_name_aliases.get(content_block.name, content_block.name)
resolved_tool_name = request_state.tool_name_aliases.get(content_block.name, content_block.name)
contents.append(
Content.from_function_call(
call_id=content_block.id,
Expand All @@ -1321,11 +1350,10 @@ def _parse_contents_from_anthropic(
)
)
case "mcp_tool_result":
call_id, _ = self._last_call_id_name or (None, None)
parsed_output: list[Content] | None = None
if content_block.content:
if isinstance(content_block.content, list):
parsed_output = self._parse_contents_from_anthropic(content_block.content)
parsed_output = self._parse_contents_from_anthropic(content_block.content, request_state)
elif isinstance(content_block.content, (str, bytes)):
parsed_output = [
Content.from_text(
Expand All @@ -1334,7 +1362,7 @@ def _parse_contents_from_anthropic(
)
]
else:
parsed_output = self._parse_contents_from_anthropic([content_block.content])
parsed_output = self._parse_contents_from_anthropic([content_block.content], request_state)
contents.append(
Content.from_mcp_server_tool_result(
call_id=content_block.tool_use_id,
Expand All @@ -1343,7 +1371,6 @@ def _parse_contents_from_anthropic(
)
)
case "web_search_tool_result" | "web_fetch_tool_result":
call_id, _ = self._last_call_id_name or (None, None)
contents.append(
Content.from_function_result(
call_id=content_block.tool_use_id,
Expand Down Expand Up @@ -1548,10 +1575,10 @@ def _parse_contents_from_anthropic(
)
case "input_json_delta":
# Skip argument deltas for MCP and server tools — execution is handled server-side.
if self._last_call_content_type in ("mcp_tool_use", "server_tool_use"):
if request_state.active_call_content_type in ("mcp_tool_use", "server_tool_use"):
pass
else:
call_id = self._last_call_id_name[0] if self._last_call_id_name else ""
call_id = request_state.active_call_id or ""
contents.append(
Content.from_function_call(
call_id=call_id,
Expand Down
Loading
Loading