Cannot Use authlib With MCP Clients

[awickham10]

Initial Checks

• I confirm that I'm using the latest version of Pydantic AI
• I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue

Description

I have several MCP servers that I need to authenticate to with an OAuth 2.0 client credentials flow. Using authlib to help handle this seems like a reasonable choice, however, I cannot get it to work.

The examples below are snippets taken from scripts - not necessarily a fully working example. I'm just trying to illustrate the patterns I've attempted.

Example Code

## Sample 1 - Using this method results in an httpx client is already opened error.


http_client = AsyncOAuth2Client(
    client_id=self.config.client_id,
    client_secret=self.config.client_secret.get_secret_value(),
    token_endpoint_auth_method="client_secret_post",
    timeout=30.0,
)

# Fetch initial token
await http_client.fetch_token(
    token_endpoint,
    grant_type="client_credentials",
    **token_kwargs,
)

server = MCPServerStreamableHTTP(
    url=target_url,
    tool_prefix=self.config.name,
    max_retries=5,
    http_client=http_client,
)


## Sample 2 - Using this method results in a missing token error.


http_client = AsyncOAuth2Client(
    client_id=self.config.client_id,
    client_secret=self.config.client_secret.get_secret_value(),
    token_endpoint_auth_method="client_secret_post",
    timeout=30.0,
)

server = MCPServerStreamableHTTP(
    url=target_url,
    tool_prefix=self.config.name,
    max_retries=5,
    http_client=http_client,
)

Python, Pydantic AI & LLM client version

python 3.12
pydantic-ai 1.2.0
openai 2.4.0

[awickham10]

I was able to get this working in an awful way by passing in an already expired token. I'd say this is still an issue though.

async def update_token_fn(token, refresh_token=None, access_token=None) -> None:
    pass

http_client = AsyncOAuth2Client(
    client_id=self.config.client_id,
    client_secret=self.config.client_secret.get_secret_value(),
    token_endpoint=token_endpoint,
    token_endpoint_auth_method="client_secret_post",
    grant_type="client_credentials",
    timeout=30.0,
    update_token=update_token_fn,
    token={
        "access_token": "fake-to-force-refresh",
        "token_type": "Bearer",
        "expires_in": -1,
    },
    **token_kwargs,
)

server = MCPServerStreamableHTTP(
    url=target_url,
    tool_prefix=self.config.name,
    max_retries=5,
    http_client=http_client,
)

[DouweM]

@awickham10 Looking at the implementation of AsyncOAuth2Client.request, I'm surprised that it doesn't seem to have a way to automatically request a token if there isn't one yet, requiring you to use the fake expired token method instead.

I think the way to go may be to use fetch_token as you did, and then to take the token and explicitly pass it in the Authentication header of a new http client, like this:

http_client = AsyncOAuth2Client(
    client_id=self.config.client_id,
    client_secret=self.config.client_secret.get_secret_value(),
    token_endpoint_auth_method="client_secret_post",
    timeout=30.0,
)

# Fetch initial token
token = await http_client.fetch_token(
    token_endpoint,
    grant_type="client_credentials",
    **token_kwargs,
)

http_client = httpx.AsyncClient(headers={'Authorization': f'Bearer {token}'})
server = MCPServerStreamableHTTP(
    url=target_url,
    tool_prefix=self.config.name,
    max_retries=5,
    http_client=http_client,
)

I would've expected your second example (no fetch_token, directly pass in the http_client) to just work, but it seems like that's not how authlib is supposed to be used. Can you see if the code above works?

[awickham10]

That does work (minor exception being I had to update the Bearer {token} to Bearer {token["access_token"]}), however I then would lose the automatic token handling that the AsyncOAuth2Client provides (automatically refresh, etc.).

P.S. Thanks for the quick response!

[DouweM]

@awickham10 I found an issue on authlib that seems to be about the same thing: authlib/authlib#531

Specifically:

Is it deliberate that in this situation, when the token can be fetched without user intervention, that an initial token must be supplied (via token keyword or a explicit call to fetch_token())?

And a similar workaround to yours:

token={'expires_at': -1, 'access_token': ''}

The issue never received a response from the maintainers though, so this looks like a legitimate authlib issue you'll have to work around, I'm afraid.

[awickham10]

Thanks so much for locating that. I'll close this one out.