mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-22 00:31:44 +00:00
78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
"""
|
|
Simple test to validate MCP auth header priority behavior.
|
|
|
|
Validates that:
|
|
1. auth_value is not required in config.yaml
|
|
2. Server-specific headers (x-mcp-server-name-authorization) take precedence over config auth_value
|
|
"""
|
|
|
|
import pytest
|
|
from litellm.proxy._experimental.mcp_server.mcp_server_manager import MCPServerManager
|
|
from litellm.types.mcp import MCPAuth, MCPTransport, MCPSpecVersion
|
|
from litellm.types.mcp_server.mcp_server_manager import MCPServer
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mcp_server_works_without_config_auth_value():
|
|
"""
|
|
Test that MCP servers work without auth_value in config when headers are provided.
|
|
This validates that auth_value is truly optional in config.yaml.
|
|
"""
|
|
# Create a server WITHOUT config auth_value
|
|
server_without_config_auth = MCPServer(
|
|
server_id="test-server-no-config",
|
|
name="Test MCP Server No Config Auth",
|
|
server_name="test_server_no_config",
|
|
alias="test_no_config",
|
|
url="https://api.example.com/mcp",
|
|
transport=MCPTransport.http,
|
|
auth_type=MCPAuth.authorization,
|
|
authentication_token=None, # No config auth
|
|
)
|
|
|
|
manager = MCPServerManager()
|
|
|
|
# Test that it works with only header auth
|
|
client = await manager._create_mcp_client(
|
|
server=server_without_config_auth,
|
|
mcp_auth_header="Bearer token_from_header_only",
|
|
)
|
|
|
|
# Verify header token is used
|
|
assert client._mcp_auth_value == "Bearer token_from_header_only"
|
|
assert client.auth_type == MCPAuth.authorization
|
|
|
|
|
|
@pytest.mark.parametrize("token_key", ["authentication_token", "auth_value"])
|
|
async def test_mcp_server_config_auth_value_header_used(token_key, config_only_mcp_manager_factory):
|
|
"""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 httpx2
|
|
|
|
from litellm.proxy._experimental.mcp_server.outbound_credentials.httpx_auth import (
|
|
StaticHeaderAuth,
|
|
)
|
|
|
|
config = {
|
|
"test_server": {
|
|
"url": "https://api.example.com/mcp",
|
|
"transport": "http",
|
|
"auth_type": "bearer_token",
|
|
token_key: "example_token",
|
|
}
|
|
}
|
|
|
|
manager = config_only_mcp_manager_factory()
|
|
await manager.load_servers_from_config(config)
|
|
|
|
server = next(iter(manager.config_mcp_servers.values()))
|
|
client = await manager._create_mcp_client(server)
|
|
|
|
assert isinstance(client._resolved_auth, StaticHeaderAuth)
|
|
emitted = next(client._resolved_auth.auth_flow(httpx2.Request("POST", server.url)))
|
|
assert emitted.headers["Authorization"] == "Bearer example_token"
|
|
assert client.auth_type == MCPAuth.bearer_token
|