test/acp/test_prompt_blocks.py

Rama main · réplica de lectura del repositorio original.

# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0

from typing import Any

import acp
import pytest
from acp import schema

from ag2 import Agent
from ag2.acp.testing import duplex_acp_config
from ag2.events import AudioInput, DocumentInput, ImageInput, TextInput
from ag2.stream import MemoryStream


class RecordingAgent:
    def __init__(self, prompts: list[list[Any]], capabilities: schema.PromptCapabilities | None = None) -> None:
        self.prompts = prompts
        self._capabilities = capabilities

    async def initialize(self, **kwargs: Any) -> schema.InitializeResponse:
        agent_capabilities = (
            schema.AgentCapabilities(prompt_capabilities=self._capabilities) if self._capabilities else None
        )
        return schema.InitializeResponse(protocol_version=acp.PROTOCOL_VERSION, agent_capabilities=agent_capabilities)

    async def new_session(self, **kwargs: Any) -> schema.NewSessionResponse:
        return schema.NewSessionResponse(session_id="recording-session")

    async def prompt(self, *, prompt: list[Any], **kwargs: Any) -> schema.PromptResponse:
        self.prompts.append(prompt)
        return schema.PromptResponse(stop_reason="end_turn")


@pytest.mark.asyncio
async def test_the_parts_of_one_message_arrive_as_separate_blocks() -> None:
    prompts: list[list[Any]] = []
    cfg = duplex_acp_config(lambda _conn: RecordingAgent(prompts))

    try:
        await Agent("acp", config=cfg).ask(TextInput("first"), TextInput("second"))
    finally:
        await cfg.aclose()

    assert prompts == [[acp.text_block("first"), acp.text_block("second")]]


async def _prompt_sent(capabilities: schema.PromptCapabilities | None, *inputs: Any) -> list[Any]:
    prompts: list[list[Any]] = []
    cfg = duplex_acp_config(lambda _conn: RecordingAgent(prompts, capabilities))

    try:
        await Agent("acp", config=cfg).ask(*inputs)
    finally:
        await cfg.aclose()

    [prompt] = prompts
    return prompt


@pytest.fixture
def all_capabilities() -> schema.PromptCapabilities:
    return schema.PromptCapabilities(image=True, audio=True, embedded_context=True)


@pytest.mark.asyncio
async def test_media_the_agent_accepts_is_sent_as_its_own_block(all_capabilities: schema.PromptCapabilities) -> None:
    prompt = await _prompt_sent(
        all_capabilities,
        TextInput("look"),
        ImageInput(data=b"img", media_type="image/png"),
        AudioInput(data=b"aud", media_type="audio/wav"),
        DocumentInput(data=b"doc", media_type="application/pdf"),
    )

    assert prompt == [
        acp.text_block("look"),
        acp.image_block("aW1n", "image/png"),
        acp.audio_block("YXVk", "audio/wav"),
        acp.resource_block(
            schema.BlobResourceContents(blob="ZG9j", mime_type="application/pdf", uri="attachment:document")
        ),
    ]


@pytest.mark.asyncio
async def test_media_the_agent_does_not_accept_is_named_not_dropped() -> None:
    prompt = await _prompt_sent(
        None,
        ImageInput(data=b"img", media_type="image/png"),
        AudioInput(data=b"aud", media_type="audio/wav"),
        DocumentInput(data=b"doc", media_type="application/pdf"),
    )

    assert prompt == [acp.text_block("[image]"), acp.text_block("[audio]"), acp.text_block("[document]")]


@pytest.mark.asyncio
async def test_a_url_goes_as_a_resource_link_whatever_the_agent_advertises() -> None:
    url = "https://example.com/photo.png"

    prompt = await _prompt_sent(None, ImageInput(url))

    assert prompt == [acp.resource_link_block(name=url, uri=url)]


@pytest.mark.asyncio
async def test_a_later_turn_sends_only_the_input_not_yet_sent() -> None:
    prompts: list[list[Any]] = []
    cfg = duplex_acp_config(lambda _conn: RecordingAgent(prompts))
    agent = Agent("acp", config=cfg)
    stream = MemoryStream()

    try:
        await agent.ask("first", stream=stream)
        await agent.ask("second", stream=stream)
    finally:
        await cfg.aclose()

    assert prompts == [[acp.text_block("first")], [acp.text_block("second")]]


@pytest.mark.asyncio
async def test_a_turn_with_no_input_still_sends_an_empty_text_block() -> None:
    assert await _prompt_sent(None) == [acp.text_block("")]