test tool limiting

examples/slackbot/src/slackbot/api.py

@@ -69,7 +69,10 @@ async def run_agent(
         counts_token = _tool_usage_counts.set(defaultdict(int))
 
         try:
-            with WatchToolCalls(settings=decorator_settings):
+            with WatchToolCalls(
+                settings=decorator_settings,
+                max_tool_calls=settings.max_tool_calls_per_turn,
+            ):
                 result = await create_agent(model=settings.model_name).run(
                     user_prompt=cleaned_message,
                     message_history=conversation,

examples/slackbot/src/slackbot/settings.py

@@ -78,20 +78,24 @@ def validate_log_level(cls, v: str) -> str:
         description="Slack user ID to notify when discussions are created (e.g., U1234567890)",
     )
 
+    # Tool use limits
+    max_tool_calls_per_turn: int = Field(
+        default=50,
+        description="Maximum number of tool calls allowed per agent turn to prevent runaway tool use",
+    )
+
     @model_validator(mode="after")
-    def validate_temperature(self) -> "SlackbotSettings":
+    def _apply_post_validation_defaults(self) -> "SlackbotSettings":
         if "gpt-5" in self.model_name:
             self.temperature = 1.0
-        return self
-
-    @model_validator(mode="after")
-    def set_turbopuffer_api_key(self) -> "SlackbotSettings":
         if not os.getenv("TURBOPUFFER_API_KEY"):
             try:
                 api_key = Secret.load("tpuf-api-key", _sync=True).get()  # type: ignore
                 os.environ["TURBOPUFFER_API_KEY"] = api_key
             except Exception:
                 pass  # If secret doesn't exist, turbopuffer will handle the error
+        if not self.admin_slack_user_id:
+            self.admin_slack_user_id = Variable.get("admin-slack-id", _sync=True)
         return self
 
     @property

examples/slackbot/src/slackbot/wrap.py

@@ -64,6 +64,7 @@ def prefect_wrapped_function(
     decorator: Callable[..., Callable[..., T]] = task,
     tags: set[str] | None = None,
     settings: dict[str, Any] | None = None,
+    max_tool_calls: int = 15,  # Default limit per agent run
 ) -> Callable[..., Callable[..., T]]:
     """Decorator for wrapping a function with a prefect decorator."""
     tags = tags or set[str]()
@@ -102,6 +103,12 @@ async def wrapper(*args, **kwargs) -> T:
                 _tool_usage_counts.set(counts)
             counts[tool_name] += 1
 
+            # Check if we've exceeded the limit
+            total_calls = sum(counts.values())
+            if total_calls > max_tool_calls:
+                # Return a message instead of raising an exception
+                return "I've reached my tool use limit for this response. Please ask a follow-up question if you need more information."
+
             # Set current tool
             _current_tool_token = _current_tool.set(tool_name)
 
@@ -161,11 +168,13 @@ def __init__(
         patch_method_name: str = "call_tool",
         tags: set[str] | None = None,
         settings: dict[str, Any] | None = None,
+        max_tool_calls: int = 15,
     ):
         """Initialize the context manager.
         Args:
             tags: Prefect tags to apply to the flow.
-            flow_kwargs: Keyword arguments to pass to the flow.
+            settings: Settings to pass to the decorator.
+            max_tool_calls: Maximum number of tool calls allowed per turn.
         """
         # Import here to avoid circular imports
         from pydantic_ai.toolsets.abstract import AbstractToolset
@@ -176,4 +185,5 @@ def __init__(
             decorator=prefect_wrapped_function,
             tags=tags,
             settings=settings,
+            max_tool_calls=max_tool_calls,
         )