Overview

azureaicommunity-agent-tool-limit is a lightweight guard layer for AI agent pipelines built on agent-framework. During each completion it tracks every function_call content item emitted by the model and silently suppresses any calls that breach a configurable global cap or an optional per-tool cap. When calls are suppressed the optional on_limit_exceeded callback is invoked.

The middleware does not raise an exception — it drops the over-limit calls so the agent loop can continue cleanly, mirroring the behaviour of the .NET ToolLimitMiddleware.

✨ Features

📦 Installation

pip install azureaicommunity-agent-tool-limit

Or install directly from source:

cd AgentFramework/Python/Middleware/ToolLimitMiddleware
pip install -e .

🚀 Quick Start

import asyncio
from agent_framework import tool
from agent_framework.ollama import OllamaChatClient
from tool_limit_middleware import ToolLimitMiddleware, ToolLimits


@tool
def get_weather(location: str) -> str:
    return f"The weather in {location} is sunny with a high of 22°C."


async def main():
    client = OllamaChatClient(model="llama3.2")

    middleware = ToolLimitMiddleware(
        limits=ToolLimits(global_max=5),
    )

    agent = client.as_agent(
        name="WeatherAgent",
        instructions="You are a helpful assistant with a weather tool.",
        tools=[get_weather],
        middleware=[middleware],
    )

    response = await agent.run("What is the weather in Amsterdam?")
    print(response.text)


asyncio.run(main())

🔧 Per-Tool Limits

In addition to the global cap, restrict individual tools independently:

middleware = ToolLimitMiddleware(
    limits=ToolLimits(
        global_max=10,
        per_tool_max={
            "get_weather":  3,
            "search_videos": 2,
        },
    ),
    on_limit_exceeded=lambda info: print("Limit hit:", info),
)

Any call to get_weather beyond 3, or to search_videos beyond 2, is silently removed — even if the global limit has not been reached.

📊 Inspect Usage

usage = middleware.get_current_usage()

print(f"Total allowed calls: {usage.total_calls} / {usage.global_limit}")

for tool_name, attempted in usage.per_tool.items():
    allowed = usage.per_tool_allowed.get(tool_name, 0)
    limit   = usage.per_tool_limits.get(tool_name)
    limit_text = f" / {limit}" if limit is not None else ""
    print(f"  {tool_name}: attempted={attempted}  allowed={allowed}{limit_text}")

# Reset for a new session
middleware.reset()

For streaming responses, a stream_transform_hook intercepts each ChatResponseUpdate as it arrives and removes over-limit function calls in real time. The on_limit_exceeded callback is fired once via a stream_result_hook after the stream completes.