	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server mTLS Authentication

MCP Server mTLS Authentication

Author: Venkata Sudhakar

Mutual TLS (mTLS) is the strongest transport-level authentication for MCP servers. Unlike one-way TLS where only the server proves its identity, mTLS requires both the server and the client to present valid certificates. This ensures that only authorised agents - those holding a certificate signed by a trusted CA - can call MCP tools.

ShopMax India uses mTLS for its internal MCP servers that handle sensitive operations like pricing updates and order approvals. Each ADK agent is provisioned with its own client certificate issued by the internal CA. The MCP server rejects any connection that cannot present a valid certificate, preventing unauthorised access even from within the private network.

The below example shows how to run an MCP server over HTTP with SSE transport secured by mTLS, using Python's ssl module to enforce mutual certificate verification.

# Install: pip install mcp aiohttp
# Generate certs: openssl req -x509 -newkey rsa:4096 -keyout ca.key -out ca.crt -days 365 -nodes
import asyncio
import ssl
from aiohttp import web
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp import types

# Certificate paths
SERVER_CERT = "certs/server.crt"
SERVER_KEY  = "certs/server.key"
CA_CERT     = "certs/ca.crt"    # CA that signed client certs

app_server = Server("shopmax-secure-mcp")

@app_server.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="approve_price_change",
            description="Approve a price change request (restricted - mTLS only)",
            inputSchema={
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"},
                    "new_price_rs": {"type": "number"},
                    "approver": {"type": "string"}
                },
                "required": ["product_id", "new_price_rs", "approver"]
            }
        )
    ]

@app_server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "approve_price_change":
        msg = ("Price change approved: product={} new_price=Rs {} by {}".format(
            arguments["product_id"],
            arguments["new_price_rs"],
            arguments["approver"]
        ))
        return [types.TextContent(type="text", text=msg)]

def create_ssl_context():
    # Server SSL context with client cert verification
    ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ctx.load_cert_chain(SERVER_CERT, SERVER_KEY)
    ctx.load_verify_locations(CA_CERT)
    ctx.verify_mode = ssl.CERT_REQUIRED   # enforce mTLS
    return ctx

async def main():
    sse = SseServerTransport("/messages")
    web_app = web.Application()

async def sse_handler(request):
        # Log the client certificate subject for audit
        peer_cert = request.transport.get_extra_info("peercert")
        if peer_cert:
            subject = dict(x[0] for x in peer_cert.get("subject", []))
            print("Client cert CN:", subject.get("commonName", "unknown"))
        async with sse.connect_sse(request.method, request.url,
                                    request.headers, request.content) as streams:
            await app_server.run(
                streams[0], streams[1],
                app_server.create_initialization_options()
            )

web_app.router.add_get("/sse", sse_handler)
    runner = web.AppRunner(web_app)
    await runner.setup()
    site = web.TCPSite(runner, "0.0.0.0", 8443, ssl_context=create_ssl_context())
    await site.start()
    print("MCP server running on https://0.0.0.0:8443 with mTLS")
    await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())

It gives the following output,

# Server starts with mTLS enforced
MCP server running on https://0.0.0.0:8443 with mTLS

# Authorised agent connects with valid cert:
Client cert CN: shopmax-pricing-agent
Tool: approve_price_change({product_id: "TV-201", new_price_rs: 59000, approver: "pricing-agent"})
Price change approved: product=TV-201 new_price=Rs 59000.0 by pricing-agent

# Unauthorised client without cert:
ssl.SSLError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed - connection rejected

Use Google Cloud Certificate Authority Service to issue and rotate client certificates for agents at scale. Store private keys in Secret Manager and mount them at runtime rather than baking them into container images. Set short certificate lifetimes (24-48 hours) and automate rotation so a compromised key has minimal exposure window. For Cloud Run deployments, combine mTLS at the application layer with VPC Service Controls at the network layer for defence in depth.

Send your comments, suggestions or queries regarding this site to [email protected].