AnthropicLlm: single_turn sub-agent turns, AgentTool dates and empty thinking blocks change the history Claude signed (2.11.0) #7480

"""Reproduce three history changes in ADK's Anthropic path (no network).

A fake Messages API behind the real AsyncAnthropic client records each
request body. For each pair of consecutive requests, the earlier
messages plus the model's reply should open the later request unchanged.
"""

import asyncio
import datetime
import json

import httpx2
from anthropic import AsyncAnthropic
from google.adk.agents import LlmAgent
from google.adk.models.anthropic_llm import AnthropicLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.agent_tool import AgentTool
from google.genai import types
from pydantic import BaseModel


class Note(BaseModel):
    day: datetime.date


def make_fake(script):
    bodies, replies = [], []

    async def handle(request):
        body = json.loads(request.content)
        if body["system"].startswith("Sub"):
            content = [{"type": "text", "text": '{"day": "2026-10-07"}'}]
            stop = "end_turn"
        else:
            content, stop = script(len(bodies))
            bodies.append(body)
            replies.append(content)
        reply = {
            "id": "msg", "type": "message", "role": "assistant",
            "model": body["model"], "content": content,
            "stop_reason": stop, "stop_sequence": None,
            "usage": {"input_tokens": 1, "output_tokens": 1},
        }
        return httpx2.Response(200, json=reply)

    client = AsyncAnthropic(
        api_key="x",
        http_client=httpx2.AsyncClient(
            transport=httpx2.MockTransport(handle)
        ),
    )
    return client, bodies, replies


def script(thinking_text):
    def turn(n):
        thinking = {
            "type": "thinking", "thinking": thinking_text,
            "signature": f"sig-{n}",
        }
        calls = [("sub", {"request": "go"}), ("clock", {})]
        if n < len(calls):
            name, args = calls[n]
            return [thinking, {
                "type": "tool_use", "id": f"toolu_{n}", "name": name,
                "input": args, "caller": {"type": "direct"},
            }], "tool_use"
        return [thinking, {"type": "text", "text": "done"}], "end_turn"

    return turn


def clock() -> dict:
    """Return the time."""
    return {"time": "noon"}


async def run(mode, thinking_text="summary", output_schema=None):
    client, bodies, replies = make_fake(script(thinking_text))
    model = AnthropicLlm(model="claude-sonnet-5-5", client=client)
    sub = LlmAgent(
        name="sub", model=model, instruction="Sub agent.",
        description="A helper.", output_schema=output_schema,
    )
    if mode == "single_turn":
        sub.mode = "single_turn"
        root = LlmAgent(name="root", model=model, instruction="Root.",
                        tools=[clock], sub_agents=[sub])
    else:
        root = LlmAgent(name="root", model=model, instruction="Root.",
                        tools=[clock, AgentTool(sub)])
    sessions = InMemorySessionService()
    runner = Runner(app_name="r", agent=root, session_service=sessions)
    await sessions.create_session(app_name="r", user_id="u", session_id="s")
    message = types.Content(role="user", parts=[types.Part(text="Hi")])
    try:
        async for _ in runner.run_async(
            user_id="u", session_id="s", new_message=message
        ):
            pass
    except Exception as exc:
        return [f"run stopped: {type(exc).__name__}: {exc}"]
    finally:
        await runner.close()
    problems = []
    for i, later in enumerate(bodies[1:]):
        # ADK also drops "caller" from tool_use blocks, ignored here.
        reply = [
            {k: v for k, v in b.items() if k != "caller"} for b in replies[i]
        ]
        sent = bodies[i]["messages"] + [
            {"role": "assistant", "content": reply}
        ]
        for n, msg in enumerate(sent):
            if n >= len(later["messages"]) or later["messages"][n] != msg:
                problems.append(
                    f"request {i + 1}, message {n}: sent "
                    f"{json.dumps(later['messages'][n])[:160]}, "
                    f"expected {json.dumps(msg)[:160]}"
                )
                break
    return problems


async def main():
    print("1. single_turn sub-agent:")
    print(*await run("single_turn"), sep="\n")
    print("2. AgentTool, output schema with a date:")
    print(*await run("agent_tool", output_schema=Note), sep="\n")
    print("3. AgentTool, empty thinking text (display omitted):")
    print(*await run("agent_tool", thinking_text=""), sep="\n")


asyncio.run(main())