Allow dynamic header creation for MCP Servers - this can be useful for authorization headers

[RRRajput]

Description

Problem

Today, when initializing pydantic_ai.mcp.MCPServerHTTP, it is possible to provide static headers that will be sent along with each request:

from pydantic_ai.mcp import MCPServerHTTP

mcp_server = MCPServerHTTP("http://localhost:8090/mcp", headers={"Authorization": "Bearer wow-much-secret"})

This is fine if your headers are always static (or if your bearer token is static). However, for some cases you might want to dynamically generate the headers before each request.

Example: You want your authorization server (Auth0 or Azure AD) to provide you with an access token that should be sent with that request. You cannot generate that token in a static manner because the tokens expire after some time and would need to be refreshed (or renewed).

Workaround

Currently, there is a work-around for this, so it is not a blocking issue.

You create the MCPServerHTTP object just before connecting to the MCP server and modify the ._mcp_servers attribute of the agent:

# agent.py
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIModel
from pydantic_ai.providers.openai import OpenAIProvider
from .constants import AGENT_MODEL_NAME, AGENT_MODEL_URL


model = OpenAIModel(
    model_name=AGENT_MODEL_NAME, provider=OpenAIProvider(base_url=AGENT_MODEL_URL)
)
my_agent = Agent(
    model=model,
    deps_type=dict[str, str],
)

# main.py
from .agent import my_agent
from .auth import get_token # Custom code
from pydantic_ai.mcp import MCPServerHTTP

async def something() -> None:
    mcp_server = MCPServerHTTP("http://localhost:8090/mcp", headers={"Authorization": get_token()})
    # The line below is the ugly part :-(
    my_agent._mcp_servers = [MCPServerHTTP(url=MCP_SERVER_URL, headers=headers)]
    async with my_agent.run_mcp_servers():
       log.info("Connected to the MCP Server. Running agent...")
       ... # Do something

Not sure how the API should be for allowing dynamically created headers, though.

References

No response

[whvneo]

With the current release of mcp python-sdk you can provide your custom httpx_client_factory.

For this we need just small changes in pydantic-ai

I created a PR for this but was struggling a bit with the typecheck but the essential commit is PR 1932

[ChuckJonas]

This is needed to be able to support multi-tenant MCP servers!

@whvneo did you ever get this working?

[fennb]

I have the same issue as this (also related: #2320).

We have implemented a workaround where we effectively create a new Agent on a per-session basis such that authentication tokens are passed right through to the MCP server (ie: against Pydantic AI conventions where Agents are effectively global singletons).

I think the main challenge here is that in Pydantic AI, MCP servers are currently agent-centric, not an agent-run-centric. This works fine if your MCP client setup is generic, but not so much if you need different MCP client setup for different runs (ie: for authentication via HTTP headers, for example).

This obviously gets more complex once you start doing stateful MCP stuff (ie: persistent streams) rather than stateless, but the overall challenge remains true, ie: you want a authenticated user session/token/whatever to be bound to an authenticated MCP request/connection.

This is sort-of feasible now (just using process_tool_call) if your MCP authentication occurs at the JSON-RPC layer, but "normal" MCP auth is done at a HTTP layer: https://modelcontextprotocol.io/specification/draft/basic/authorization#access-token-usage

There's a few different ways I can think of for tackling this:

1. HTTP Client Factory
Introduce the concept of an HttpxClientFactory (or similar) that can setup the HTTP client used for the MCP server connection on a per-anything basis (as per here) - ie: specify a factory that can create the MCP Httpx client based on runtime context.

2. process_tool_call approach
Intercept/mutate the tool calls before their made to the MCP server as per here, but extend process_tool_call to be able to modify the HTTP request itself (somehow). This presents potential problems with stateful connections.

3. Make MCP client parameters run-centric (ish)
Previously, before agent.run_mcp_servers() was deprecated, this could have potentially been a pattern, ie:

async with agent.run_mcp_servers(http_client=create_mcp_http_client(headers={'Authorization': 'blah'}):
    result = await agent.run("Profit")

(or even some variant where you pull from a pool of of connections or similar).

However, given this approach is now deprecated in favour of async with agent: we now don't have anywhere to pass in the arguments. I'm not sure the right answer here (nor am I familiar with the target patterns the Pydantic AI team prefers/is aiming for).

Regardless, the statefulness of MCP client-server connections presents challenges given that a connection with authorization (ie: session) should be able to persist outside of the run-context, which really means the run-centric approach doesn't quite work in isolation (but nor does the agent-centric one).

Related: It is likely useful for a server to be able to adjust the tools/list output based on authorization and have the MCP client "do the right thing". I only mention this as it may help inform where the auth should happen.

As mentioned, our solution so far is to just create a custom Agent configuration for every session (with a per-user MCP server), but it seems less than ideal given the way agents are supposed to be setup/used.

Related slack thread here.

[DouweM]

@fennb Thanks for sharing your findings! Another workaround that I haven't verified but would expect to work would be to build a custom httpx.AsyncClient subclass that overrides build_request to inject additional headers (possibly using some arbitrary function you pass into your custom class's initializer), and then to pass that to the MCPServer as http_client.

We're also considering a more native way to build toolsets (including MCP servers) at agent run time, possibly based on agent run context, in #2331 -- please have a look and weigh in there as well!

[fennb]

Now that #2361 is merged, this should be possible using agent.run(toolsets=...) as of v0.4.10.

An MCP server is a toolset now, so you should be able to do stuff like (to modify the original example in this thread):

def do_something(auth_token):
    mcp_server = MCPServerHTTP("http://localhost:8090/mcp", headers={"Authorization": auth_token})
    result = await agent.run(toolsets=[mcp_server])

If you want persistent/stateful connections to multiple MCP servers, it should be possible to create/manage a pool of them yourself and just re-use them as required.

I haven't had time to actually test this (I will soon), but based on the recent v0.4.10 code changes, it should work.

[fennb]

I haven't had time to actually test this (I will soon), but based on the recent v0.4.10 code changes, it should work.

I can confirm that this approach works well and solve the specific problem mentioned in the issue. In my case, I actually implemented a server connection pool (keyed by user session) so that user sessions/MCP server connection instances are bound 1:1 (even though this isn't strictly necessary if using stateless MCP).

[DouweM]

Since multiple workarounds are already available, and #2331 will provide a native way to define a toolset whose config is dependent on the run context, I'll close this as a duplicate of that.