feat: support tool-specific call limits

libs/agno/agno/agent/agent.py

@@ -281,6 +281,8 @@ class Agent:
     # Add a tool that allows the Model to search the knowledge base (aka Agentic RAG)
     # Added only if knowledge is provided.
     search_knowledge: bool = True
+    # Call limit for search_knowledge tool per run. None means no limit.
+    search_knowledge_call_limit: Optional[int] = None
     # Add a tool that allows the Agent to update Knowledge.
     update_knowledge: bool = False
     # Add a tool that allows the Model to get the tool call history.
@@ -467,6 +469,7 @@ def __init__(
         reasoning_max_steps: int = 10,
         read_chat_history: bool = False,
         search_knowledge: bool = True,
+        search_knowledge_call_limit: Optional[int] = None,
         update_knowledge: bool = False,
         read_tool_call_history: bool = False,
         send_media_to_model: bool = True,
@@ -586,6 +589,7 @@ def __init__(
 
         self.read_chat_history = read_chat_history
         self.search_knowledge = search_knowledge
+        self.search_knowledge_call_limit = search_knowledge_call_limit
         self.update_knowledge = update_knowledge
         self.read_tool_call_history = read_tool_call_history
         self.send_media_to_model = send_media_to_model
@@ -9540,7 +9544,10 @@ async def asearch_knowledge_base(query: str) -> str:
         else:
             search_knowledge_base_function = search_knowledge_base  # type: ignore
 
-        return Function.from_callable(search_knowledge_base_function, name="search_knowledge_base")
+        search_func = Function.from_callable(search_knowledge_base_function, name="search_knowledge_base")
+        if self.search_knowledge_call_limit is not None:
+            search_func.call_limit = self.search_knowledge_call_limit
+        return search_func
 
     def _search_knowledge_base_with_agentic_filters_function(
         self,
@@ -9623,10 +9630,13 @@ async def asearch_knowledge_base(query: str, filters: Optional[List[KnowledgeFil
         else:
             search_knowledge_base_function = search_knowledge_base  # type: ignore
 
-        return Function.from_callable(
+        search_func = Function.from_callable(
             search_knowledge_base_function,
             name="search_knowledge_base_with_agentic_filters",
         )
+        if self.search_knowledge_call_limit is not None:
+            search_func.call_limit = self.search_knowledge_call_limit
+        return search_func
 
     def add_to_knowledge(self, query: str, result: str) -> str:
         """Use this function to add information to the knowledge base for future use.

libs/agno/agno/models/base.py

@@ -301,6 +301,26 @@ def _format_tools(self, tools: Optional[List[Union[Function, dict]]]) -> List[Di
                 _tool_dicts.append(tool)
         return _tool_dicts
 
+    def _extract_tool_call_limits(self, tools: Optional[List[Union[Function, dict]]]) -> Dict[str, int]:
+        """
+        Extract the call limits for each tool from the provided tool list.
+
+        Args:
+            tools: List of tools, each of which may have a single call limit.
+
+        Returns:
+            A dictionary mapping tool names to their respective call limits, or an empty dictionary if no limits are set.
+        """
+        if not tools:
+            return {}
+
+        tool_limits: Dict[str, int] = {}
+        for tool in tools:
+            if isinstance(tool, Function) and tool.call_limit is not None:
+                tool_limits[tool.name] = tool.call_limit
+
+        return tool_limits
+
     def response(
         self,
         messages: List[Message],
@@ -340,6 +360,7 @@ def response(
         model_response = ModelResponse()
 
         function_call_count = 0
+        remaining_tool_limits = self._extract_tool_call_limits(tools)
 
         _tool_dicts = self._format_tools(tools) if tools is not None else []
         _functions = {tool.name: tool for tool in tools if isinstance(tool, Function)} if tools is not None else {}
@@ -380,6 +401,7 @@ def response(
                     function_call_results=function_call_results,
                     current_function_call_count=function_call_count,
                     function_call_limit=tool_call_limit,
+                    remaining_tool_limits=remaining_tool_limits,
                 ):
                     if isinstance(function_call_response, ModelResponse):
                         # The session state is updated by the function call
@@ -505,7 +527,7 @@ async def aresponse(
 
         _tool_dicts = self._format_tools(tools) if tools is not None else []
         _functions = {tool.name: tool for tool in tools if isinstance(tool, Function)} if tools is not None else {}
-
+        remaining_tool_limits = self._extract_tool_call_limits(tools)
         function_call_count = 0
 
         while True:
@@ -544,6 +566,7 @@ async def aresponse(
                     function_call_results=function_call_results,
                     current_function_call_count=function_call_count,
                     function_call_limit=tool_call_limit,
+                    remaining_tool_limits=remaining_tool_limits,
                 ):
                     if isinstance(function_call_response, ModelResponse):
                         # The session state is updated by the function call
@@ -885,7 +908,7 @@ def response_stream(
 
         _tool_dicts = self._format_tools(tools) if tools is not None else []
         _functions = {tool.name: tool for tool in tools if isinstance(tool, Function)} if tools is not None else {}
-
+        remaining_tool_limits = self._extract_tool_call_limits(tools)
         function_call_count = 0
 
         while True:
@@ -955,6 +978,7 @@ def response_stream(
                     function_call_results=function_call_results,
                     current_function_call_count=function_call_count,
                     function_call_limit=tool_call_limit,
+                    remaining_tool_limits=remaining_tool_limits,
                 ):
                     if self.cache_response and isinstance(function_call_response, ModelResponse):
                         streaming_responses.append(function_call_response)
@@ -1084,7 +1108,7 @@ async def aresponse_stream(
 
         _tool_dicts = self._format_tools(tools) if tools is not None else []
         _functions = {tool.name: tool for tool in tools if isinstance(tool, Function)} if tools is not None else {}
-
+        remaining_tool_limits = self._extract_tool_call_limits(tools)
         function_call_count = 0
 
         while True:
@@ -1153,6 +1177,7 @@ async def aresponse_stream(
                     function_call_results=function_call_results,
                     current_function_call_count=function_call_count,
                     function_call_limit=tool_call_limit,
+                    remaining_tool_limits=remaining_tool_limits,
                 ):
                     if self.cache_response and isinstance(function_call_response, ModelResponse):
                         streaming_responses.append(function_call_response)
@@ -1389,10 +1414,17 @@ def create_function_call_result(
             **kwargs,  # type: ignore
         )
 
-    def create_tool_call_limit_error_result(self, function_call: FunctionCall) -> Message:
+    def create_tool_call_limit_error_result(
+        self, function_call: FunctionCall, is_individual_tool: bool = False
+    ) -> Message:
+        if is_individual_tool:
+            content = f"Call limit reached for {function_call.function.name}. Don't try to execute this tool again."
+        else:
+            content = f"Tool call limit reached. Tool call {function_call.function.name} not executed. Don't try to execute it again."
+
         return Message(
             role=self.tool_message_role,
-            content=f"Tool call limit reached. Tool call {function_call.function.name} not executed. Don't try to execute it again.",
+            content=content,
             tool_call_id=function_call.call_id,
             tool_name=function_call.function.name,
             tool_args=function_call.arguments,
@@ -1543,6 +1575,7 @@ def run_function_calls(
         additional_input: Optional[List[Message]] = None,
         current_function_call_count: int = 0,
         function_call_limit: Optional[int] = None,
+        remaining_tool_limits: Optional[Dict[str, int]] = None,
     ) -> Iterator[Union[ModelResponse, RunOutputEvent, TeamRunOutputEvent]]:
         # Additional messages from function calls that will be added to the function call results
         if additional_input is None:
@@ -1556,6 +1589,12 @@ def run_function_calls(
                     function_call_results.append(self.create_tool_call_limit_error_result(fc))
                     continue
 
+            # Check if individual tool has exceeded its call limit
+            if remaining_tool_limits and fc.function.name in remaining_tool_limits:
+                if remaining_tool_limits[fc.function.name] <= 0:
+                    function_call_results.append(self.create_tool_call_limit_error_result(fc, is_individual_tool=True))
+                    continue
+                remaining_tool_limits[fc.function.name] -= 1
             paused_tool_executions = []
 
             # The function cannot be executed without user confirmation
@@ -1687,6 +1726,7 @@ async def arun_function_calls(
         current_function_call_count: int = 0,
         function_call_limit: Optional[int] = None,
         skip_pause_check: bool = False,
+        remaining_tool_limits: Optional[Dict[str, int]] = None,
     ) -> AsyncIterator[Union[ModelResponse, RunOutputEvent, TeamRunOutputEvent]]:
         # Additional messages from function calls that will be added to the function call results
         if additional_input is None:
@@ -1701,6 +1741,13 @@ async def arun_function_calls(
                     function_call_results.append(self.create_tool_call_limit_error_result(fc))
                     # Skip this function call
                     continue
+
+            # Check if individual tool has exceeded its call limit
+            if remaining_tool_limits and fc.function.name in remaining_tool_limits:
+                if remaining_tool_limits[fc.function.name] <= 0:
+                    function_call_results.append(self.create_tool_call_limit_error_result(fc, is_individual_tool=True))
+                    continue
+                remaining_tool_limits[fc.function.name] -= 1
             function_calls_to_run.append(fc)
 
         # Yield tool_call_started events for all function calls or pause them

libs/agno/agno/tools/decorator.py

@@ -76,6 +76,7 @@ def tool(
     cache_results: bool = False,
     cache_dir: Optional[str] = None,
     cache_ttl: int = 3600,
+    call_limit: Optional[int] = None,
 ) -> Callable[[F], Function]: ...
 
 
@@ -104,6 +105,7 @@ def tool(*args, **kwargs) -> Union[Function, Callable[[F], Function]]:
         cache_results: bool - If True, enable caching of function results
         cache_dir: Optional[str] - Directory to store cache files
         cache_ttl: int - Time-to-live for cached results in seconds
+        call_limit: Optional[int] - The maximum number of calls allowed during a single run. None means no limit.
 
     Returns:
         Union[Function, Callable[[F], Function]]: Decorated function or decorator
@@ -141,6 +143,7 @@ async def my_async_function():
             "cache_results",
             "cache_dir",
             "cache_ttl",
+            "call_limit",
         }
     )
 
@@ -229,6 +232,7 @@ async def async_gen_wrapper(*args: Any, **kwargs: Any) -> Any:
             "cache_results": kwargs.get("cache_results", False),
             "cache_dir": kwargs.get("cache_dir"),
             "cache_ttl": kwargs.get("cache_ttl", 3600),
+            "call_limit": kwargs.get("call_limit"),
             **{
                 k: v
                 for k, v in kwargs.items()
@@ -241,6 +245,7 @@ async def async_gen_wrapper(*args: Any, **kwargs: Any) -> Any:
                     "cache_results",
                     "cache_dir",
                     "cache_ttl",
+                    "call_limit",
                 ]
                 and v is not None
             },

libs/agno/agno/tools/function.py

@@ -118,6 +118,9 @@ class Function(BaseModel):
     cache_dir: Optional[str] = None
     cache_ttl: int = 3600
 
+    # The maximum number of calls allowed during a single run. None means no limit.
+    call_limit: Optional[int] = None
+
     # --*-- FOR INTERNAL USE ONLY --*--
     # The agent that the function is associated with
     _agent: Optional[Any] = None