scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-09 03:18:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 109

test(mcp): assert config token via resolved_auth, not the headers dict

Browse source 
test_mcp_server_config_auth_value_header_used inspected _get_auth_headers(), but the
graft now carries the static credential on the client's httpx.Auth (resolved_auth) and
writes the header at send time, so that dict is empty. Assert the header the
StaticHeaderAuth emits onto the request instead. Both config keys (authentication_token,
auth_value) stay covered.
This commit is contained in:
Tin Chi Lo 2026-06-23 19:24:43 -07:00
parent aa6d27c181
commit ed4c55df07
1 changed files with 15 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
tests/mcp_tests/test_mcp_auth_priority.py
View file

@ -45,7 +45,18 @@ async def test_mcp_server_works_without_config_auth_value():
@pytest.mark.parametrize("token_key", ["authentication_token", "auth_value"])
async def test_mcp_server_config_auth_value_header_used(token_key):
"""Ensure auth header is sent when auth token configured in config"""
"""Ensure the configured auth token is emitted as the upstream Authorization header.
The token is resolved through the v2 credential resolver and rides on the client's
httpx.Auth, so assert the header it writes onto the request rather than the (now
credential-free) _get_auth_headers() dict.
"""
import httpx
from litellm.proxy._experimental.mcp_server.outbound_credentials.httpx_auth import (
StaticHeaderAuth,
)
config = {
"test_server": {
"url": "https://api.example.com/mcp",
@ -60,7 +71,8 @@ async def test_mcp_server_config_auth_value_header_used(token_key):
server = next(iter(manager.config_mcp_servers.values()))
client = await manager._create_mcp_client(server)
headers = client._get_auth_headers()
assert headers["Authorization"] == "Bearer example_token"
assert isinstance(client._resolved_auth, StaticHeaderAuth)
emitted = next(client._resolved_auth.auth_flow(httpx.Request("POST", server.url)))
assert emitted.headers["Authorization"] == "Bearer example_token"
assert client.auth_type == MCPAuth.bearer_token