Python: Fix tool execution bleed-over in aiohttp/Bot Framework scenarios (#2314)

* Deep copy the agent chat options to avoid mutations

* avoiding _thread.RLock pickling errors

python/packages/core/tests/core/test_agents.py

@@ -115,6 +115,26 @@ async def test_chat_client_agent_prepare_thread_and_messages(chat_client: ChatCl
     assert result_messages[1].text == "Test"
 
 
+async def test_prepare_thread_does_not_mutate_agent_chat_options(chat_client: ChatClientProtocol) -> None:
+    tool = HostedCodeInterpreterTool()
+    agent = ChatAgent(chat_client=chat_client, tools=[tool])
+
+    assert agent.chat_options.tools is not None
+    base_tools = agent.chat_options.tools
+    thread = agent.get_new_thread()
+
+    _, prepared_chat_options, _ = await agent._prepare_thread_and_messages(  # type: ignore[reportPrivateUsage]
+        thread=thread,
+        input_messages=[ChatMessage(role=Role.USER, text="Test")],
+    )
+
+    assert prepared_chat_options.tools is not None
+    assert base_tools is not prepared_chat_options.tools
+
+    prepared_chat_options.tools.append(HostedCodeInterpreterTool())  # type: ignore[arg-type]
+    assert len(agent.chat_options.tools) == 1
+
+
 async def test_chat_client_agent_update_thread_id(chat_client_base: ChatClientProtocol) -> None:
     mock_response = ChatResponse(
         messages=[ChatMessage(role=Role.ASSISTANT, contents=[TextContent("test response")])],

python/packages/core/tests/core/test_function_invocation_logic.py

@@ -3,6 +3,7 @@
 import pytest
 
 from agent_framework import (
+    ChatAgent,
     ChatClientProtocol,
     ChatMessage,
     ChatOptions,
@@ -127,6 +128,148 @@ async def test_base_client_with_streaming_function_calling(chat_client_base: Cha
     assert exec_counter == 1
 
 
+async def test_function_invocation_inside_aiohttp_server(chat_client_base: ChatClientProtocol):
+    import aiohttp
+    from aiohttp import web
+
+    exec_counter = 0
+
+    @ai_function(name="start_todo_investigation")
+    def ai_func(user_query: str) -> str:
+        nonlocal exec_counter
+        exec_counter += 1
+        return f"Investigated {user_query}"
+
+    chat_client_base.run_responses = [
+        ChatResponse(
+            messages=ChatMessage(
+                role="assistant",
+                contents=[
+                    FunctionCallContent(
+                        call_id="1",
+                        name="start_todo_investigation",
+                        arguments='{"user_query": "issue"}',
+                    )
+                ],
+            )
+        ),
+        ChatResponse(messages=ChatMessage(role="assistant", text="done")),
+    ]
+
+    agent = ChatAgent(chat_client=chat_client_base, tools=[ai_func])
+
+    async def handler(request: web.Request) -> web.Response:
+        thread = agent.get_new_thread()
+        result = await agent.run("Fix issue", thread=thread)
+        return web.Response(text=result.text or "")
+
+    app = web.Application()
+    app.add_routes([web.post("/run", handler)])
+
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.TCPSite(runner, "127.0.0.1", 0)
+    await site.start()
+    try:
+        port = site._server.sockets[0].getsockname()[1]
+        async with aiohttp.ClientSession() as session, session.post(f"http://127.0.0.1:{port}/run") as response:
+            assert response.status == 200
+            await response.text()
+    finally:
+        await runner.cleanup()
+
+    assert exec_counter == 1
+
+
+async def test_function_invocation_in_threaded_aiohttp_app(chat_client_base: ChatClientProtocol):
+    import asyncio
+    import threading
+    from queue import Queue
+
+    import aiohttp
+    from aiohttp import web
+
+    exec_counter = 0
+
+    @ai_function(name="start_threaded_investigation")
+    def ai_func(user_query: str) -> str:
+        nonlocal exec_counter
+        exec_counter += 1
+        return f"Threaded {user_query}"
+
+    chat_client_base.run_responses = [
+        ChatResponse(
+            messages=ChatMessage(
+                role="assistant",
+                contents=[
+                    FunctionCallContent(
+                        call_id="thread-1",
+                        name="start_threaded_investigation",
+                        arguments='{"user_query": "issue"}',
+                    )
+                ],
+            )
+        ),
+        ChatResponse(messages=ChatMessage(role="assistant", text="done")),
+    ]
+
+    agent = ChatAgent(chat_client=chat_client_base, tools=[ai_func])
+
+    ready_event = threading.Event()
+    port_queue: Queue[int] = Queue()
+    shutdown_queue: Queue[tuple[asyncio.AbstractEventLoop, asyncio.Event]] = Queue()
+
+    async def init_app() -> web.Application:
+        async def handler(request: web.Request) -> web.Response:
+            thread = agent.get_new_thread()
+            result = await agent.run("Fix issue", thread=thread)
+            return web.Response(text=result.text or "")
+
+        app = web.Application()
+        app.add_routes([web.post("/run", handler)])
+        return app
+
+    def server_thread() -> None:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
+        async def runner_main() -> None:
+            app = await init_app()
+            runner = web.AppRunner(app)
+            await runner.setup()
+            site = web.TCPSite(runner, "127.0.0.1", 0)
+            await site.start()
+            shutdown_event = asyncio.Event()
+            shutdown_queue.put((loop, shutdown_event))
+            port = site._server.sockets[0].getsockname()[1]
+            port_queue.put(port)
+            ready_event.set()
+            try:
+                await shutdown_event.wait()
+            finally:
+                await runner.cleanup()
+
+        try:
+            loop.run_until_complete(runner_main())
+        finally:
+            loop.close()
+
+    thread = threading.Thread(target=server_thread, daemon=True)
+    thread.start()
+    ready_event.wait(timeout=5)
+    assert ready_event.is_set()
+    loop_ref, shutdown_event = shutdown_queue.get(timeout=2)
+    port = port_queue.get(timeout=2)
+
+    async with aiohttp.ClientSession() as session, session.post(f"http://127.0.0.1:{port}/run") as response:
+        assert response.status == 200
+        await response.text()
+
+    loop_ref.call_soon_threadsafe(shutdown_event.set)
+    thread.join(timeout=5)
+    assert exec_counter == 1
+
+
 @pytest.mark.parametrize(
     "approval_required,num_functions",
     [