diff --git a/python/packages/copilotstudio/agent_framework_copilotstudio/_agent.py b/python/packages/copilotstudio/agent_framework_copilotstudio/_agent.py index 0bc9c484327..1f7c61a5874 100644 --- a/python/packages/copilotstudio/agent_framework_copilotstudio/_agent.py +++ b/python/packages/copilotstudio/agent_framework_copilotstudio/_agent.py @@ -5,6 +5,7 @@ from collections.abc import AsyncIterable, Awaitable, Sequence from typing import Any, Literal, TypedDict, overload +import aiohttp from agent_framework import ( AgentMiddlewareTypes, AgentResponse, @@ -20,11 +21,53 @@ from agent_framework._settings import load_settings from agent_framework._types import AgentRunInputs from agent_framework.exceptions import AgentException +from microsoft_agents.activity import Activity, ActivityTypes from microsoft_agents.copilotstudio.client import AgentType, ConnectionSettings, CopilotClient, PowerPlatformCloud from ._acquire_token import acquire_token +async def _patched_post_request( + self: CopilotClient, url: str, data: dict[str, Any], headers: dict[str, Any] +) -> AsyncIterable[Activity]: + async with aiohttp.ClientSession() as session, session.post(url, json=data, headers=headers) as response: + if response.status != 200: + detail = await response.text() + raise AgentException(f"Copilot Studio request to {url} failed ({response.status}): {detail}") + event_type = None + buffer = b"" + async for chunk in response.content.iter_any(): + buffer += chunk + while b"\n" in buffer: + line, buffer = buffer.split(b"\n", 1) + if line.startswith(b"event:"): + event_type = line[6:].decode("utf-8").strip() + if line.startswith(b"data:") and event_type == "activity": + activity_data = line[5:].decode("utf-8").strip() + activity = Activity.model_validate_json(activity_data) + + if activity.type == ActivityTypes.message: + self._current_conversation_id = activity.conversation.id # pyright: ignore[reportPrivateUsage] + + yield activity + if buffer: + line = buffer + if line.startswith(b"event:"): + event_type = line[6:].decode("utf-8").strip() + if line.startswith(b"data:") and event_type == "activity": + activity_data = line[5:].decode("utf-8").strip() + activity = Activity.model_validate_json(activity_data) + + if activity.type == ActivityTypes.message: + self._current_conversation_id = activity.conversation.id # pyright: ignore[reportPrivateUsage] + + yield activity + + +# Monkeypatch CopilotClient to handle data lines larger than 512KB without LineTooLong errors. +CopilotClient.post_request = _patched_post_request + + class CopilotStudioSettings(TypedDict, total=False): """Copilot Studio model settings. diff --git a/python/packages/copilotstudio/tests/test_line_too_long_bug.py b/python/packages/copilotstudio/tests/test_line_too_long_bug.py new file mode 100644 index 00000000000..95af89359a9 --- /dev/null +++ b/python/packages/copilotstudio/tests/test_line_too_long_bug.py @@ -0,0 +1,65 @@ +# Copyright (c) Microsoft. All rights reserved. + +from collections.abc import AsyncIterable + +import pytest +from aiohttp import web +from microsoft_agents.activity import Activity +from microsoft_agents.copilotstudio.client import ConnectionSettings, CopilotClient + +import agent_framework_copilotstudio # noqa: F401 + +# Side-effect import: applies CopilotClient.post_request monkeypatch. + + +async def mock_server_handler(request: web.Request) -> web.StreamResponse: + headers = {"Content-Type": "text/event-stream"} + response = web.StreamResponse(headers=headers) + await response.prepare(request) + + # 600KB line (larger than aiohttp default limit of 524288 bytes) + long_data = "x" * 600000 + + # Write event type line + await response.write(b"event: activity\n") + # Write data line + data_line = f'data: {{"type":"message","text":"{long_data}","conversation":{{"id":"test"}}}}\n'.encode() + await response.write(data_line) + return response + + +@pytest.fixture +async def local_server() -> AsyncIterable[str]: + app = web.Application() + app.router.add_post("/test", mock_server_handler) + runner = web.AppRunner(app) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + + address = runner.addresses[0] + port = address[1] + # Yield the URL to the test + yield f"http://127.0.0.1:{port}/test" + + await site.stop() + await runner.cleanup() + + +@pytest.mark.asyncio +async def test_copilot_client_post_request_long_line(local_server: str) -> None: + settings = ConnectionSettings( + environment_id="env", + agent_identifier="agent", + cloud=None, + copilot_agent_type=None, + custom_power_platform_cloud=None, + ) + client = CopilotClient(settings=settings, token="token") + + activities: list[Activity] = [] + async for activity in client.post_request(local_server, {}, {}): + activities.append(activity) + + assert len(activities) == 1 + assert activities[0].text == "x" * 600000 diff --git a/python/packages/core/tests/core/test_function_invocation_logic.py b/python/packages/core/tests/core/test_function_invocation_logic.py index ae01d82a226..b950aab0f06 100644 --- a/python/packages/core/tests/core/test_function_invocation_logic.py +++ b/python/packages/core/tests/core/test_function_invocation_logic.py @@ -2374,6 +2374,7 @@ def test_replace_approval_contents_with_results_allows_reused_call_id_after_comp ("call_reused", "second output"), ] + def test_replace_approval_contents_with_results_uses_result_call_ids_for_placeholders() -> None: from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results