Dynamic Headers

Setting dynamic headers with Agno MCP tools

When using MCP tools, you will often want to send information to the MCP server via HTTP headers. For example, you may want to pass information about the current user, or authorization related data.

To do this dynamically, you can use the header_provider parameter when initializing the MCPTools class.

You will also be able to access the RunContext object, which contains the current run’s context, including the user ID, session ID, and metadata, together with the contextual Agent or Team. Simply add the run_context, agent, and team parameters to your header_provider function.

from agno.tools.mcp import MCPTools

# Our header_provider function, which will generate the headers per run.
def header_provider(
    run_context: RunContext, # The current run's context
    agent: Optional["Agent"] = None, # The contextual Agent instance
    team: Optional["Team"] = None, # The contextual Team instance
) -> dict:
    headers = {
        "X-User-ID": run_context.user_id or "unknown",
        "X-Session-ID": run_context.session_id or "unknown",
        "X-Run-ID": run_context.run_id,
        "X-Agent-Name": agent.name if agent else None,
        "X-Team-Name": team.name if team else None,
    }
    return headers

# When these MCP tools are used, the header_provider function will be used to update the headers for each run.
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

The header_provider function you pass to your MCPTools instance will be used to update the headers every run.

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	Contains run-specific data like run_id, user_id, session_id, and metadata
agent	Agent	The Agent instance making the tool call (if applicable)
team	Team	The Team instance making the tool call (if applicable)

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

Complete Example

1. 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)

2. Run the example client:

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: RunContext) -> dict:
    """Generate headers from the current run context."""
    return {
        "X-User-ID": run_context.user_id or "anonymous",
        "X-Session-ID": run_context.session_id or "no-session",
        "X-Run-ID": run_context.run_id,
    }


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

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

Usage with MultiMCPTools

Dynamic headers work the same way with MultiMCPTools:

from agno.tools.mcp import MultiMCPTools

mcp_tools = MultiMCPTools(
    urls=[
        "http://server1.example.com/mcp",
        "http://server2.example.com/mcp",
    ],
    transport="streamable-http",
    header_provider=header_provider,  # Applied to all servers
)