Update

Author: dirkbrnd

cookbook/agent_concepts/user_control_flows/external_tool_execution_async_responses.py

@@ -27,7 +27,12 @@ def execute_shell_command(command: str) -> str:
     Returns:
         str: The output of the shell command
     """
-    if command.startswith("ls ") or command == "ls" or command.startswith("cat ") or command.startswith("head "):
+    if (
+        command.startswith("ls ")
+        or command == "ls"
+        or command.startswith("cat ")
+        or command.startswith("head ")
+    ):
         return subprocess.check_output(command, shell=True).decode("utf-8")
     raise Exception(f"Unsupported command: {command}")
 
@@ -41,7 +46,9 @@ def execute_shell_command(command: str) -> str:
 run_response = asyncio.run(agent.arun("What files do I have in my current directory?"))
 
 # Keep executing externally-required tools until the run completes
-while run_response.is_paused and len(run_response.tools_awaiting_external_execution) > 0:
+while (
+    run_response.is_paused and len(run_response.tools_awaiting_external_execution) > 0
+):
     for tool in run_response.tools_awaiting_external_execution:
         if tool.tool_name == execute_shell_command.name:
             print(f"Executing {tool.tool_name} with args {tool.tool_args} externally")
@@ -57,5 +64,3 @@ def execute_shell_command(command: str) -> str:
 
 # Or for simple debug flow
 # agent.print_response("What files do I have in my current directory?")
-
-

libs/agno/agno/models/aws/bedrock.py

@@ -31,6 +31,7 @@
 BEDROCK_SUPPORTED_VIDEO_FORMATS = ["mp4", "mov", "mkv", "webm", "flv", "mpeg", "mpg", "wmv", "three_gp"]
 BEDROCK_SUPPORTED_FILE_FORMATS = ["pdf", "csv", "doc", "docx", "xls", "xlsx", "html", "txt", "md"]
 
+
 @dataclass
 class AwsBedrock(Model):
     """

libs/agno/tests/integration/models/openai/responses/test_tool_use.py

@@ -5,7 +5,6 @@
 
 from agno.agent import Agent, RunResponse  # noqa
 from agno.models.openai import OpenAIResponses
-from agno.tools.duckduckgo import DuckDuckGoTools
 from agno.tools.exa import ExaTools
 from agno.tools.yfinance import YFinanceTools
 
@@ -153,6 +152,7 @@ def test_parallel_tool_calls():
 
 def test_multiple_tool_calls():
     """Test multiple different tool types with the responses API."""
+
     def get_the_weather(city: str):
         return f"It is currently 70 degrees and cloudy in {city}"
 

libs/agno/tests/unit/models/openai/test_openai_responses_id_handling.py

@@ -1,9 +1,9 @@
 from typing import Any, Dict, List, Optional
 
+from agno.models.base import MessageData
 from agno.models.message import Message
 from agno.models.openai.responses import OpenAIResponses
 from agno.models.response import ModelResponse
-from agno.models.base import MessageData
 
 
 class _FakeError:
@@ -47,7 +47,15 @@ def __init__(self, *, _id: str, call_id: Optional[str], name: str, arguments: st
 
 
 class _FakeStreamEvent:
-    def __init__(self, *, type: str, item: Optional[_FakeStreamItem] = None, delta: str = "", response: Any = None, annotation: Any = None):
+    def __init__(
+        self,
+        *,
+        type: str,
+        item: Optional[_FakeStreamItem] = None,
+        delta: str = "",
+        response: Any = None,
+        annotation: Any = None,
+    ):
         self.type = type
         self.item = item
         self.delta = delta
@@ -66,15 +74,22 @@ def test_format_messages_maps_tool_output_fc_to_call_id():
                 "id": "fc_abc123",
                 "call_id": "call_def456",
                 "type": "function",
-                "function": {"name": "execute_shell_command", "arguments": "{\"command\": \"ls -la\"}"},
+                "function": {"name": "execute_shell_command", "arguments": '{"command": "ls -la"}'},
             }
         ],
     )
 
     # Tool output referring to the fc_* id should be normalized to call_*
     tool_output = Message(role="tool", tool_call_id="fc_abc123", content="ok")
 
-    fm = model._format_messages(messages=[Message(role="system", content="s"), Message(role="user", content="u"), assistant_with_tool_call, tool_output])
+    fm = model._format_messages(
+        messages=[
+            Message(role="system", content="s"),
+            Message(role="user", content="u"),
+            assistant_with_tool_call,
+            tool_output,
+        ]
+    )
 
     # Expect one function_call and one function_call_output normalized
     fc_items = [x for x in fm if x.get("type") == "function_call"]
@@ -114,7 +129,10 @@ def test_process_stream_response_builds_tool_calls():
     stream_data = MessageData()
 
     # Simulate function_call added and then completed
-    added = _FakeStreamEvent(type="response.output_item.added", item=_FakeStreamItem(_id="fc_abc123", call_id="call_def456", name="execute", arguments="{}"))
+    added = _FakeStreamEvent(
+        type="response.output_item.added",
+        item=_FakeStreamItem(_id="fc_abc123", call_id="call_def456", name="execute", arguments="{}"),
+    )
     mr, tool_use = model._process_stream_response(added, assistant_message, stream_data, {})
     assert mr is None
 
@@ -155,9 +173,14 @@ def test_reasoning_previous_response_skips_prior_function_call_items(monkeypatch
         ],
     )
 
-    fm = model._format_messages(messages=[Message(role="system", content="s"), Message(role="user", content="u"), assistant_with_prev, assistant_with_tool_call])
+    fm = model._format_messages(
+        messages=[
+            Message(role="system", content="s"),
+            Message(role="user", content="u"),
+            assistant_with_prev,
+            assistant_with_tool_call,
+        ]
+    )
 
     # Expect no re-sent function_call when previous_response_id is present for reasoning models
     assert all(x.get("type") != "function_call" for x in fm)
-
-