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. Reject missing or unauthorized user IDs. 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

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)

Run the example client:

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())

You can check the logs of your MCP server to see the headers that were sent.

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])

Get Started

Basics

Advanced

Production

Providers

Other

Additional Resources

Dynamic Headers

The `header_provider` function

Complete Example

Multiple servers

​The header_provider function

​Complete Example

​Multiple servers

The `header_provider` function

Complete Example

Multiple servers