feat: Add auth-headers cluster provider for dynamic multi-tenant Kubernetes access

[mfaizanse]

This PR introduces a new auth-headers cluster provider strategy that enables multi-tenant Kubernetes MCP server deployments where cluster connection details and authentication credentials are provided via HTTP headers on a per-request basis.

Summary

The auth-headers provider allows clients to connect to different Kubernetes clusters dynamically without server-side configuration or stored credentials. All cluster connection details (server URL, CA certificate, authentication) are provided in request headers, enabling true zero-trust and multi-tenant architectures.

Documentation

Complete documentation added in docs/AUTH_HEADERS_PROVIDER.md.

Custom Headers

Clients must provide these headers with each request:

Required:

• kubernetes-server - Kubernetes API server URL
• kubernetes-certificate-authority-data - Base64-encoded CA certificate

Authentication (one required):

• kubernetes-authorization - Bearer token, OR
• kubernetes-client-certificate-data + kubernetes-client-key-data - Client cert/key pair

Optional:

• kubernetes-insecure-skip-tls-verify - Set to true to skip TLS verification

Backward Compatibility

✅ Fully backward compatible - existing providers unchanged
✅ No breaking changes to existing APIs
✅ New provider is opt-in via configuration

Example usage (Python):

import asyncio
import httpx
import json
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport

def get_k8s_headers(cluster_json_file, authType):
    # authType: "token" or "cert"

    with open(cluster_json_file) as f:
      target_cluster = json.load(f)

    headers = {
      "kubernetes-certificate-authority-data": target_cluster["TEST_CLUSTER_CA_DATA"],
      "kubernetes-server": target_cluster["TEST_CLUSTER_URL"]
    }

    if authType == "token":
      headers["kubernetes-authorization"] = target_cluster["TEST_CLUSTER_AUTH_TOKEN"]
    elif authType == "cert":
      headers["kubernetes-client-certificate-data"] = target_cluster["TEST_CLUSTER_CLIENT_CERTIFICATE_DATA"]
      headers["kubernetes-client-key-data"] = target_cluster["TEST_CLUSTER_CLIENT_KEY_DATA"]
    return headers



async def connect_and_list_pods():    
    server_url = "http://localhost:8085/mcp"
    cluster_json_file = "k8s1.json"
    authType = "cert"
    
    # Configure custom headers if needed for the Kubernetes MCP server
    custom_headers = get_k8s_headers(cluster_json_file, authType)    
    try:
        # Define the tool parameters
        tool_name = "pods_list_in_namespace"
        tool_args = {
            "namespace": "default"
        }
        
        async with Client(
            transport=StreamableHttpTransport(
                server_url, 
                headers=custom_headers,
            ),
        ) as client:
            response = await client.call_tool(tool_name, tool_args, meta=custom_headers)
            print(response)           
    except httpx.ConnectError:
        print(f"✗ Failed to connect to {server_url}")
    except Exception as e:
        print(f"✗ Error: {type(e).__name__}: {e}")
        import traceback
        traceback.print_exc()


def main():
    """Main entry point."""
    asyncio.run(connect_and_list_pods())


if __name__ == "__main__":
    main()

Related Issues

• [RFE] specify cluster details on request #387

[mfaizanse]

@manusa any update on the review?

[manusa]

Hi @mfaizanse

I haven't had the chance to check this one yet. Sorry.

One of the main concerns we had while evaluating #387 is what are the implications of switching clusters in a single chat completion session:

Switch clusters per request without changing tool parameters

It'd be great if you could elaborate on specific use cases or scenarios where this provider would be useful.