Dynamic Headers

Use header_provider to calculate HTTP headers when MCPTools connects and before each run.

The initial connection and tool discovery happen before a run exists. run_context, agent, and team can all be None on that first call. A protected MCP server therefore needs a credential that works during discovery. The example below uses one least-privileged service identity that is safe for every caller.

import os
from typing import Optional

from agno.run import RunContext
from agno.tools.mcp import MCPTools

service_token = os.environ["MCP_SERVICE_TOKEN"]


def header_provider(
    run_context: Optional[RunContext] = None,
    agent: Optional["Agent"] = None,
    team: Optional["Team"] = None,
) -> dict:
    headers = {
        "Authorization": f"Bearer {service_token}",
        "X-User-ID": (run_context.user_id if run_context else None) or "unknown",
        "X-Session-ID": (run_context.session_id if run_context else None) or "unknown",
        "X-Run-ID": (run_context.run_id if run_context else None) or "unknown",
        "X-Agent-Name": agent.name if agent else "unknown",
        "X-Team-Name": team.name if team else "unknown",
    }
    return headers

mcp_tools = MCPTools(
    url="http://localhost:8000/mcp",
    header_provider=header_provider,
)

Dynamic headers are only relevant when using HTTP-based transports (streamable-http or sse). The stdio transport does not support headers.

The `header_provider` function

MCPTools calls the provider during initial connection and updates the headers before each run. Handle the initial call without relying on run-specific values.

The service token above is sent during discovery and every run. Give it only permissions shared by all callers. For per-user authorization, use run_context.user_id as a non-secret lookup key and resolve the credential from an external secrets store inside the provider. Enforce authorization before starting the run/tool request or on the MCP server. Exceptions from header_provider are caught and treated as empty dynamic headers; static headers can remain in use. Raising from this callback alone does not deny access. Do not place bearer tokens in RunContext.metadata; run metadata is serialized and can be persisted with the session.

The function is expected to return a dict of header name-value pairs.

The following parameters will be automatically injected into the function and can be useful to generate the headers:

Parameter	Type	Description
`run_context`	`RunContext`	Current run data such as `run_id`, `user_id`, `session_id`, and `metadata`. `None` during initial connection
`agent`	`Agent`	Agent making the tool call. `None` during initial connection or for a Team
`team`	`Team`	Team making the tool call. `None` during initial connection or for an Agent

You can read more about the RunContext object and its fields in the RunContext reference.

Complete Example

Create and activate a Python environment, then install the dependencies:

uv pip install -U "agno[mcp,openai]"
export OPENAI_API_KEY="your_openai_api_key"

This local server displays request headers; it does not authenticate callers. The earlier service-token fragment assumes a separately protected server and MCP_SERVICE_TOKEN set in its client environment.

Run the example MCP server:

server.py

from fastmcp import FastMCP
from fastmcp.server import Context
from fastmcp.server.dependencies import get_http_request

mcp = FastMCP("My Server")


@mcp.tool
async def greet(name: str, ctx: Context) -> str:
    """Greet a user with personalized information from headers."""
    # Get the HTTP request object
    request = get_http_request()

    # Access headers (lowercase!)
    user_id = request.headers.get("x-user-id", "unknown")
    tenant_id = request.headers.get("x-tenant-id", "unknown")
    agent_name = request.headers.get("x-agent-name", "unknown")

    print("=" * 60)
    print(f"Headers -> Agent: {agent_name}, User: {user_id}, Tenant: {tenant_id}")
    print("=" * 60)

    return f"Hello, {name}! (User: {user_id}, Tenant: {tenant_id})"


if __name__ == "__main__":
    mcp.run(transport="streamable-http", port=8000)

Save the server as server.py and run python server.py. Keep it running.

Save the example client as client.py:

import asyncio
from typing import Optional

from agno.agent import Agent
from agno.models.openai import OpenAIResponses
from agno.run import RunContext
from agno.tools.mcp import MCPTools


def header_provider(run_context: Optional[RunContext] = None) -> dict:
    """Generate headers from the current run context."""
    return {
        "X-User-ID": (run_context.user_id if run_context else None) or "anonymous",
        "X-Session-ID": (run_context.session_id if run_context else None) or "no-session",
        "X-Run-ID": (run_context.run_id if run_context else None) or "unknown",
    }


async def main():
    # Create MCPTools with dynamic headers
    mcp_tools = MCPTools(
        url="http://localhost:8000/mcp",
        transport="streamable-http",
        header_provider=header_provider,  # Enable dynamic headers
    )
    await mcp_tools.connect()

    try:
        agent = Agent(
            model=OpenAIResponses(id="gpt-5.2"),
            tools=[mcp_tools],
        )

        # The header_provider receives context from these parameters
        await agent.arun(
            "Hello, my name is Bob!",
            user_id="user-123",
            session_id="session-456",
        )
    finally:
        await mcp_tools.close()

if __name__ == "__main__":
    asyncio.run(main())

In a second terminal with the same environment activated and key configured, run python client.py. Inspect the MCP server logs to see the headers.

Multiple servers

Pass the same provider to each MCPTools instance:

from agno.agent import Agent
from agno.tools.mcp import MCPTools

server_one_tools = MCPTools(
    url="http://server1.example.com/mcp",
    header_provider=header_provider,
)
server_two_tools = MCPTools(
    url="http://server2.example.com/mcp",
    header_provider=header_provider,
)

agent = Agent(tools=[server_one_tools, server_two_tools])

The header_provider function

Complete Example

Multiple servers

On this page

The `header_provider` function