Python: additional Foundry Tools (#611)

* initial work on additional foundry tools

* fixes

* fix tests

* fix import

* updated lock

* added hosted MCP for foundry

* fixes

* fix for test

* updated samples

* fix result parsing

python/samples/getting_started/agents/foundry/foundry_with_code_interpreter.py

@@ -2,35 +2,30 @@
 
 import asyncio
 
-from agent_framework import AgentRunResponseUpdate, ChatAgent, ChatResponseUpdate, HostedCodeInterpreterTool
-from agent_framework.foundry import FoundryChatClient
-from azure.ai.agents.models import (
-    RunStepDelta,
-    RunStepDeltaChunk,
-    RunStepDeltaCodeInterpreterDetailItemObject,
-    RunStepDeltaCodeInterpreterToolCall,
-    RunStepDeltaToolCallObject,
+from agent_framework import (
+    AgentRunResponse,
+    HostedCodeInterpreterTool,
 )
+from agent_framework.foundry import FoundryChatClient
 from azure.identity.aio import AzureCliCredential
 
 
-def get_code_interpreter_chunk(chunk: AgentRunResponseUpdate) -> str | None:
+def print_code_interpreter_inputs(response: AgentRunResponse) -> None:
     """Helper method to access code interpreter data."""
-    if (
-        isinstance(chunk.raw_representation, ChatResponseUpdate)
-        and isinstance(chunk.raw_representation.raw_representation, RunStepDeltaChunk)
-        and isinstance(chunk.raw_representation.raw_representation.delta, RunStepDelta)
-        and isinstance(chunk.raw_representation.raw_representation.delta.step_details, RunStepDeltaToolCallObject)
-        and chunk.raw_representation.raw_representation.delta.step_details.tool_calls
-    ):
-        for tool_call in chunk.raw_representation.raw_representation.delta.step_details.tool_calls:
-            if (
-                isinstance(tool_call, RunStepDeltaCodeInterpreterToolCall)
-                and isinstance(tool_call.code_interpreter, RunStepDeltaCodeInterpreterDetailItemObject)
-                and tool_call.code_interpreter.input is not None
-            ):
-                return tool_call.code_interpreter.input
-    return None
+    from agent_framework import ChatResponseUpdate
+    from azure.ai.agents.models import (
+        RunStepDeltaCodeInterpreterDetailItemObject,
+    )
+
+    print("\nCode Interpreter Inputs during the run:")
+    if response.raw_representation is None:
+        return
+    for chunk in response.raw_representation:
+        if isinstance(chunk, ChatResponseUpdate) and isinstance(
+            chunk.raw_representation, RunStepDeltaCodeInterpreterDetailItemObject
+        ):
+            print(chunk.raw_representation.input, end="")
+    print("\n")
 
 
 async def main() -> None:
@@ -41,24 +36,19 @@ async def main() -> None:
     # authentication option.
     async with (
         AzureCliCredential() as credential,
-        ChatAgent(
-            chat_client=FoundryChatClient(async_credential=credential),
+        FoundryChatClient(async_credential=credential) as chat_client,
+    ):
+        agent = chat_client.create_agent(
+            name="CodingAgent",
             instructions="You are a helpful assistant that can write and execute Python code to solve problems.",
             tools=HostedCodeInterpreterTool(),
-        ) as agent,
-    ):
-        query = "Generate the factorial of 100 using python code."
+        )
+        query = "Generate the factorial of 100 using python code, show the code and execute it."
         print(f"User: {query}")
-        print("Agent: ", end="", flush=True)
-        generated_code = ""
-        async for chunk in agent.run_stream(query):
-            if chunk.text:
-                print(chunk.text, end="", flush=True)
-            code_interpreter_chunk = get_code_interpreter_chunk(chunk)
-            if code_interpreter_chunk is not None:
-                generated_code += code_interpreter_chunk
-
-        print(f"\nGenerated code:\n{generated_code}")
+        response = await AgentRunResponse.from_agent_response_generator(agent.run_stream(query))
+        print(f"Agent: {response}")
+        # To review the code interpreter outputs, you can access them from the response raw_representations, just uncomment the next line:
+        # print_code_interpreter_inputs(response)
 
 
 if __name__ == "__main__":

python/samples/getting_started/agents/foundry/foundry_with_hosted_mcp.py

@@ -0,0 +1,65 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from typing import Any
+
+from agent_framework import AgentProtocol, AgentThread, HostedMCPTool
+from agent_framework.foundry import FoundryChatClient
+from azure.identity.aio import AzureCliCredential
+
+
+async def handle_approvals_with_thread(query: str, agent: "AgentProtocol", thread: "AgentThread"):
+    """Here we let the thread deal with the previous responses, and we just rerun with the approval."""
+    from agent_framework import ChatMessage
+
+    result = await agent.run(query, thread=thread, store=True)
+    while len(result.user_input_requests) > 0:
+        new_input: list[Any] = []
+        for user_input_needed in result.user_input_requests:
+            print(
+                f"User Input Request for function from {agent.name}: {user_input_needed.function_call.name}"
+                f" with arguments: {user_input_needed.function_call.arguments}"
+            )
+            user_approval = input("Approve function call? (y/n): ")
+            new_input.append(
+                ChatMessage(
+                    role="user",
+                    contents=[user_input_needed.create_response(user_approval.lower() == "y")],
+                )
+            )
+        result = await agent.run(new_input, thread=thread, store=True)
+    return result
+
+
+async def main() -> None:
+    """Example showing Hosted MCP tools for a Foundry Agent."""
+    async with (
+        AzureCliCredential() as credential,
+        FoundryChatClient(async_credential=credential) as chat_client,
+    ):
+        # enable foundry observability
+        await chat_client.setup_foundry_observability()
+        agent = chat_client.create_agent(
+            name="DocsAgent",
+            instructions="You are a helpful assistant that can help with microsoft documentation questions.",
+            tools=HostedMCPTool(
+                name="Microsoft Learn MCP",
+                url="https://learn.microsoft.com/api/mcp",
+            ),
+        )
+        thread = agent.get_new_thread()
+        # First query
+        query1 = "How to create an Azure storage account using az cli?"
+        print(f"User: {query1}")
+        result1 = await handle_approvals_with_thread(query1, agent, thread)
+        print(f"{agent.name}: {result1}\n")
+        print("\n=======================================\n")
+        # Second query
+        query2 = "What is Microsoft Semantic Kernel?"
+        print(f"User: {query2}")
+        result2 = await handle_approvals_with_thread(query2, agent, thread)
+        print(f"{agent.name}: {result2}\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

python/samples/getting_started/agents/foundry/foundry_with_multiple_tools.py

@@ -0,0 +1,82 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from datetime import datetime, timezone
+from typing import Any
+
+from agent_framework import (
+    AgentProtocol,
+    AgentThread,
+    HostedMCPTool,
+    HostedWebSearchTool,
+)
+from agent_framework.foundry import FoundryChatClient
+from azure.identity.aio import AzureCliCredential
+
+
+def get_time() -> str:
+    """Get the current UTC time."""
+    current_time = datetime.now(timezone.utc)
+    return f"The current UTC time is {current_time.strftime('%Y-%m-%d %H:%M:%S')}."
+
+
+async def handle_approvals_with_thread(query: str, agent: "AgentProtocol", thread: "AgentThread"):
+    """Here we let the thread deal with the previous responses, and we just rerun with the approval."""
+    from agent_framework import ChatMessage
+
+    result = await agent.run(query, thread=thread, store=True)
+    while len(result.user_input_requests) > 0:
+        new_input: list[Any] = []
+        for user_input_needed in result.user_input_requests:
+            print(
+                f"User Input Request for function from {agent.name}: {user_input_needed.function_call.name}"
+                f" with arguments: {user_input_needed.function_call.arguments}"
+            )
+            user_approval = input("Approve function call? (y/n): ")
+            new_input.append(
+                ChatMessage(
+                    role="user",
+                    contents=[user_input_needed.create_response(user_approval.lower() == "y")],
+                )
+            )
+        result = await agent.run(new_input, thread=thread, store=True)
+    return result
+
+
+async def main() -> None:
+    """Example showing Hosted MCP tools for a Foundry Agent."""
+    async with (
+        AzureCliCredential() as credential,
+        FoundryChatClient(async_credential=credential) as chat_client,
+    ):
+        # enable foundry observability
+        await chat_client.setup_foundry_observability()
+        agent = chat_client.create_agent(
+            name="DocsAgent",
+            instructions="You are a helpful assistant that can help with microsoft documentation questions.",
+            tools=[
+                HostedMCPTool(
+                    name="Microsoft Learn MCP",
+                    url="https://learn.microsoft.com/api/mcp",
+                ),
+                # needs BING_CONNECTION_ID set in the env
+                HostedWebSearchTool(count=5),
+                get_time,
+            ],
+        )
+        thread = agent.get_new_thread()
+        # First query
+        query1 = "How to create an Azure storage account using az cli and what time is it?"
+        print(f"User: {query1}")
+        result1 = await handle_approvals_with_thread(query1, agent, thread)
+        print(f"{agent.name}: {result1}\n")
+        print("\n=======================================\n")
+        # Second query
+        query2 = "What is Microsoft Semantic Kernel and use a web search to see what is Reddit saying about it?"
+        print(f"User: {query2}")
+        result2 = await handle_approvals_with_thread(query2, agent, thread)
+        print(f"{agent.name}: {result2}\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())