How to Integrate Headroom with Agno: Optimize Token Usage in AI Agents

Wrap any Agno model with HeadroomAgnoModel to automatically compress context and reduce LLM token costs without modifying your agent logic.

Headroom provides a first-class integration for the Agno AI-agent framework that inserts token-optimization pipelines directly into your LLM calls. By wrapping your existing Agno models with HeadroomAgnoModel, you can achieve significant context compression while maintaining full compatibility with Agno's logging, tool loops, and streaming capabilities. This integration resides in the chopratejas/headroom repository and offers drop-in support for both synchronous and asynchronous workflows.

Installation and Setup

Install Headroom with Agno support using pip:

# Install Headroom with Agno extras

pip install "headroom-ai[agno]"

# Install Agno framework (if not already present)

pip install agno

Core Integration Components

The integration architecture consists of three primary components that work together to provide seamless token optimization.

Model Wrapper (HeadroomAgnoModel)

The HeadroomAgnoModel class in headroom/integrations/agno/model.py inherits from agno.models.base.Model and serves as a transparent wrapper around any Agno model. It intercepts all LLM calls—including invoke, ainvoke, and invoke_stream—applies Headroom's TransformPipeline for context compression, and forwards the optimized requests to the underlying provider.

The wrapper maintains a thread-safe metrics history and tracks a running total of tokens saved across all requests. It automatically converts Agno Message objects to the OpenAI-style dictionary format required by Headroom's optimization engine, then converts the results back to preserve Agno's internal messaging system. Extended-thinking blocks used by Claude are preserved untouched to ensure provider compatibility.

Provider Detection (get_headroom_provider)

Located in headroom/integrations/agno/providers.py, the get_headroom_provider function automatically inspects the wrapped Agno model's class name, module path, or model ID to select the appropriate Headroom token-counting backend. This supports major providers including OpenAI, Anthropic, Google, and Cohere, ensuring accurate token estimation regardless of your underlying LLM.

Observability Hooks (HeadroomPreHook and HeadroomPostHook)

The optional hooks defined in headroom/integrations/agno/hooks.py expose detailed token-saving metrics and alerting capabilities. HeadroomPreHook executes before optimization, while HeadroomPostHook can emit alerts when requests exceed specified token thresholds. These hooks integrate with Agno's pre- and post-hook system to provide real-time observability into your optimization savings.

Basic Usage Examples

Wrap an Agno Model for Immediate Optimization

Use HeadroomAgnoModel to wrap any existing Agno model with zero changes to your agent logic:

from agno.agent import Agent
from agno.models.openai import OpenAIChat
from headroom.integrations.agno import HeadroomAgnoModel

# Wrap any Agno model

model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))

# Use with a normal Agno agent

agent = Agent(model=model)
response = agent.run("What is the capital of France?")
print(response)
print(f"Tokens saved: {model.total_tokens_saved}")

Adding Observability Hooks

Monitor token usage and configure alerts using the hook system:

from headroom.integrations.agno import (
    HeadroomAgnoModel,
    HeadroomPreHook,
    HeadroomPostHook,
    create_headroom_hooks,
)

model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))

# Option 1: Manual instantiation

pre_hook = HeadroomPreHook()
post_hook = HeadroomPostHook(token_alert_threshold=10_000)

# Option 2: Convenience factory

pre_hook, post_hook = create_headroom_hooks(
    token_alert_threshold=5_000,
    log_level="DEBUG",
)

agent = Agent(
    model=model,
    pre_hooks=[pre_hook],
    post_hooks=[post_hook],
)

# Run requests

agent.run("Summarize the latest AI news.")
print(f"Total tokens saved: {model.total_tokens_saved}")
print("Summary:", post_hook.get_summary())

Async Usage for High-Throughput Applications

The wrapper fully supports Agno's async methods for concurrent processing:

import asyncio
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from headroom.integrations.agno import HeadroomAgnoModel

async def main():
    model = HeadroomAgnoModel(OpenAIChat(id="gpt-4o"))
    agent = Agent(model=model)

    # Async response

    resp = await agent.aresponse(["user", "Explain quantum tunnelling."])
    print(resp)

    # Async streaming

    async for chunk in await agent.aresponse_stream(["user", "Give me a story."]):
        print(chunk, end="", flush=True)

asyncio.run(main())

Stand-Alone Optimization Without Agents

Use the optimize_messages utility directly when you don't need a full Agent:

from headroom.integrations.agno import optimize_messages
from agno.models.openai import OpenAIChat

messages = [
    {"role": "system", "content": "You are a helpful assistant."},
    {"role": "user", "content": "Analyse this huge JSON payload..."},
]

opt_msgs, metrics = optimize_messages(
    messages,
    model="gpt-4o",  # For token estimation

)
print(f"Saved {metrics['tokens_saved']} tokens")

Architecture Deep Dive

According to the chopratejas/headroom source code, the integration preserves Agno's internal mechanics by converting messages to OpenAI-style dictionaries before applying Headroom's compression algorithms. The headroom/integrations/agno/model.py file handles method overrides for all Agno Model base class methods, ensuring that tool loops, function calling, and streaming responses continue to function normally.

The provider detection system in headroom/integrations/agno/providers.py eliminates manual configuration by mapping Agno model classes to Headroom's token-counting backends. This allows the optimization pipeline to accurately estimate token counts for both the compressed request and the original, ensuring accurate savings calculations.

Summary

• HeadroomAgnoModel wraps any Agno model to insert token optimization before each LLM call
• Automatic provider detection supports OpenAI, Anthropic, Google, and Cohere without manual configuration
• Thread-safe metrics track total tokens saved across all requests
• Observability hooks provide real-time alerting and detailed usage summaries
• Full async support includes ainvoke, aresponse, and streaming methods
• Zero code changes required to existing Agno agents—simply wrap the model

Frequently Asked Questions

How do I integrate Headroom with Agno without changing my existing agent code?

Wrap your existing Agno model with HeadroomAgnoModel before passing it to the Agent constructor. The wrapper intercepts all LLM calls transparently, so your agent's behavior, tool usage, and logging remain unchanged while automatically benefiting from context compression.

Does Headroom support async operations with Agno?

Yes, the HeadroomAgnoModel class in headroom/integrations/agno/model.py overrides Agno's async methods including ainvoke and aresponse_stream. You can use standard await syntax and async iterators exactly as you would with unwrapped Agno models.

Which LLM providers are supported for token counting?

The provider detection system in headroom/integrations/agno/providers.py automatically configures token counting for OpenAI, Anthropic, Google, and Cohere models based on the wrapped model's class name and module path. This ensures accurate token estimation regardless of your backend provider.

Can I use Headroom's optimization without creating an Agno Agent?

Yes, import the optimize_messages function from headroom.integrations.agno to apply Headroom's TransformPipeline directly to message dictionaries. This returns optimized messages and metrics without requiring an Agent instance, useful for standalone LLM calls.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}