feat(auth): 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.

[mfaizanse]

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.

Hi @manusa,

Thanks for the response!

We have a multi-tenant k8s agent which receives requests from frontend along with k8s auth headers. The user authentication (OAuth etc.) is handled in frontend. We want to connect the agent with the MCP server, but in this case we cannot use kubeconfig based provider. The conversation session is maintained in the agent application and the MCP would be used as stateless.

I see this PR was merged to allow custom managers, but for header based provider, we also need the propagation middleware.

P.S. I will resolve the merge conflict after the review so I rebased to the latest main once.

[schoeppi5]

Hey @mfaizanse,

awesome work on this feature. I was looking into implementing something similar and found your PR.
We have a very similar use case to what you are trying to achieve and this feature would help us a great deal!

I am part of a platform team maintaining a fleet of clusters. We want to offer our users MCP access to their clusters. Some have access to only one cluster, some to multiple. Access management is done with OIDC. This feature would allow users to access the clusters they have with exactly the rights they have.

Let me know, if there is something I can help out with.

_{Philipp Schöppner <[email protected]>, Mercedes-Benz Tech Innovation GmbH (Provider Information)}

[mfaizanse]

Hi @manusa,
Is there any status update on this topic?