Dynamic Headers

Setting dynamic headers with Kern 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.

1from kern.tools.mcp import MCPTools
2
3# Our header_provider function, which will generate the headers per run.
4def header_provider(
5    run_context: RunContext, # The current run's context
6    agent: Optional["Agent"] = None, # The contextual Agent instance
7    team: Optional["Team"] = None, # The contextual Team instance
8) -> dict:
9    headers = {
10        "X-User-ID": run_context.user_id or "unknown",
11        "X-Session-ID": run_context.session_id or "unknown",
12        "X-Run-ID": run_context.run_id,
13        "X-Agent-Name": agent.name if agent else None,
14        "X-Team-Name": team.name if team else None,
15    }
16    return headers
17
18# When these MCP tools are used, the header_provider function will be used to update the headers for each run.
19mcp_tools = MCPTools(
20    url="http://localhost:8000/mcp",
21    header_provider=header_provider,
22)

Note

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

Run the example MCP server:

1from fastmcp import FastMCP
2from fastmcp.server import Context
3from fastmcp.server.dependencies import get_http_request
4
5mcp = FastMCP("My Server")
6
7
8@mcp.tool
9async def greet(name: str, ctx: Context) -> str:
10    """Greet a user with personalized information from headers."""
11    # Get the HTTP request object
12    request = get_http_request()
13
14    # Access headers (lowercase!)
15    user_id = request.headers.get("x-user-id", "unknown")
16    tenant_id = request.headers.get("x-tenant-id", "unknown")
17    agent_name = request.headers.get("x-agent-name", "unknown")
18
19    print("=" * 60)
20    print(f"Headers -> Agent: {agent_name}, User: {user_id}, Tenant: {tenant_id}")
21    print("=" * 60)
22
23    return f"Hello, {name}! (User: {user_id}, Tenant: {tenant_id})"
24
25
26if __name__ == "__main__":
27    mcp.run(transport="streamable-http", port=8000)

Run the example client:

1from kern.agent import Agent
2from kern.models.openai import OpenAIResponses
3from kern.run import RunContext
4from kern.tools.mcp import MCPTools
5
6
7def header_provider(run_context: RunContext) -> dict:
8    """Generate headers from the current run context."""
9    return {
10        "X-User-ID": run_context.user_id or "anonymous",
11        "X-Session-ID": run_context.session_id or "no-session",
12        "X-Run-ID": run_context.run_id,
13    }
14
15
16async def main():
17    # Create MCPTools with dynamic headers
18    mcp_tools = MCPTools(
19        url="http://localhost:8000/mcp",
20        transport="streamable-http",
21        header_provider=header_provider,  # Enable dynamic headers
22    )
23    await mcp_tools.connect()
24
25    try:
26        agent = Agent(
27            model=OpenAIResponses(id="gpt-5.2"),
28            tools=[mcp_tools],
29        )
30
31        # The header_provider receives context from these parameters
32        await agent.arun(
33            "Hello, my name is Bob!",
34            user_id="user-123",
35            session_id="session-456",
36        )
37    finally:
38        await mcp_tools.close()
39
40if __name__ == "__main__":
41    asyncio.run(main())

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:

1from kern.tools.mcp import MultiMCPTools
2
3mcp_tools = MultiMCPTools(
4    urls=[
5        "http://server1.example.com/mcp",
6        "http://server2.example.com/mcp",
7    ],
8    transport="streamable-http",
9    header_provider=header_provider,  # Applied to all servers
10)

Dynamic Headers

The header_provider function

Complete Example

Usage with MultiMCPTools

The `header_provider` function