test_a2a_hitl.py

"""A2A example: human-in-the-loop email approval flow.

Sends a message that triggers your_send_email_tool (which is configured with
HumanInTheLoopMiddleware), waits for the interrupt, then resumes by approving
the tool call.

Usage:
    export LANGSMITH_API_KEY=lsv2_...
    uv run python test_a2a_hitl.py
"""

import asyncio
import os

import httpx
from a2a.client import ClientConfig, ClientFactory
from a2a.types import DataPart, Message, Part, Role, TextPart

AGENT_URL = "https://<deployment-url>/a2a/<assistant-id>"
HEADERS = {"X-Api-Key": os.environ["LANGSMITH_API_KEY"]}


async def main():
    async with httpx.AsyncClient(headers=HEADERS) as httpx_client:
        client = await ClientFactory.connect(
            AGENT_URL,
            client_config=ClientConfig(streaming=False, httpx_client=httpx_client),
            resolver_http_kwargs={"headers": HEADERS},
        )

        # Step 1: Send a message that triggers the email tool (will interrupt)
        async for task, event in client.send_message(
            Message(
                message_id="send-email",
                parts=[
                    Part(
                        root=TextPart(
                            text="Send an email to bob@example.com with subject 'Hello' and body 'Hi Bob!'"
                        )
                    )
                ],
                role=Role.user,
            )
        ):
            print("Interrupted — waiting for approval")
            print(task)
            print()

        # Step 2: Resume by approving the interrupted tool call
        async for task, event in client.send_message(
            Message(
                message_id="approve-email",
                context_id=task.context_id,
                task_id=task.id,
                parts=[
                    Part(
                        root=DataPart(
                            data={"resume": {"decisions": [{"type": "approve"}]}}
                        )
                    )
                ],
                role=Role.user,
            )
        ):
            print("Approved — task completed")
            print(task)


asyncio.run(main())