Service-to-Service Authentication for Multi-Agent Systems: mTLS and Service Tokens

Why Service-to-Service Auth Matters in Multi-Agent Systems

In a multi-agent architecture, agents communicate with each other to delegate tasks, share context, and coordinate workflows. A triage agent routes requests to specialized agents. A research agent calls a tool-execution agent. An orchestrator fans out work to multiple worker agents. Each of these interactions is an internal API call that needs authentication.

Without service-to-service authentication, a compromised or malicious service can impersonate any other service in the system. An attacker who gains access to one agent can make requests to every other agent as if they were legitimate. This is why zero-trust architecture principles demand that every internal call is authenticated, regardless of network boundaries.

Two Approaches: mTLS vs Service Tokens

Mutual TLS (mTLS) — both the client and server present X.509 certificates. The connection itself is authenticated at the transport layer. The server verifies the client's certificate before any application code runs. This provides strong cryptographic identity and is the standard in service meshes like Istio and Linkerd.

flowchart TD
    INPUT(["Task input"])
    SUPER["Supervisor agent<br/>plans plus monitors"]
    W1["Worker 1<br/>research"]
    W2["Worker 2<br/>code"]
    W3["Worker 3<br/>writing"]
    CRITIC{"Output meets<br/>rubric?"}
    REWORK["Rework or<br/>retry path"]
    SHARED[("Shared scratchpad<br/>and memory")]
    OUT(["Final result"])
    INPUT --> SUPER
    SUPER --> W1 --> CRITIC
    SUPER --> W2 --> CRITIC
    SUPER --> W3 --> CRITIC
    W1 --> SHARED
    W2 --> SHARED
    W3 --> SHARED
    SHARED --> SUPER
    CRITIC -->|Pass| OUT
    CRITIC -->|Fail| REWORK --> SUPER
    style SUPER fill:#4f46e5,stroke:#4338ca,color:#fff
    style CRITIC fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OUT fill:#059669,stroke:#047857,color:#fff
    style SHARED fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b

Service Tokens — each service has a pre-shared secret or retrieves a short-lived token from a central authority. The token is passed in the Authorization header. This is simpler to implement but requires careful secret management and token rotation.

In practice, production multi-agent systems use both: mTLS at the transport layer for mutual authentication, and service tokens at the application layer for authorization and scope enforcement.

Setting Up mTLS for Agent Communication

First, generate a CA (Certificate Authority) and service certificates. In production, use a tool like cert-manager, Vault PKI, or your cloud provider's CA service:

# Generate CA key and certificate
openssl genrsa -out ca.key 4096
openssl req -new -x509 -key ca.key -sha256 -subj "/CN=AgentCA" \
    -days 365 -out ca.crt

# Generate a key and CSR for the research-agent service
openssl genrsa -out research-agent.key 2048
openssl req -new -key research-agent.key \
    -subj "/CN=research-agent" -out research-agent.csr

# Sign the certificate with the CA
openssl x509 -req -in research-agent.csr -CA ca.crt -CAkey ca.key \
    -CAcreateserial -out research-agent.crt -days 90 -sha256

Configuring FastAPI for mTLS

Use uvicorn's SSL parameters to require client certificates:

# run_server.py
import uvicorn

if __name__ == "__main__":
    uvicorn.run(
        "main:app",
        host="0.0.0.0",
        port=8443,
        ssl_keyfile="certs/research-agent.key",
        ssl_certfile="certs/research-agent.crt",
        ssl_ca_certs="certs/ca.crt",
        # Require client certificate — this enables mTLS
        ssl_cert_reqs=2,  # ssl.CERT_REQUIRED
    )

Extract the client identity from the TLS connection in middleware:

from fastapi import Request, HTTPException

class ServiceIdentity:
    def __init__(self, service_name: str, common_name: str):
        self.service_name = service_name
        self.common_name = common_name

async def get_service_identity(request: Request) -> ServiceIdentity:
    """Extract client certificate CN from mTLS connection."""
    client_cert = request.scope.get("transport", {})

    # In production with a reverse proxy, the cert info
    # is typically passed via headers
    client_cn = request.headers.get("X-Client-CN")
    if not client_cn:
        raise HTTPException(status_code=401, detail="No client certificate")

    return ServiceIdentity(
        service_name=client_cn,
        common_name=client_cn,
    )

Service Token Implementation

For environments where mTLS is complex to set up (local development, certain cloud platforms), service tokens provide a pragmatic alternative. Each agent service has a unique token that identifies it:

# auth/service_auth.py
import os
import secrets
import hashlib
from datetime import datetime, timezone, timedelta
from fastapi import Depends, HTTPException, Header

# Service registry — in production, use Vault or a secrets manager
SERVICE_SECRETS = {
    "triage-agent": os.environ.get("TRIAGE_AGENT_SECRET"),
    "research-agent": os.environ.get("RESEARCH_AGENT_SECRET"),
    "tool-executor": os.environ.get("TOOL_EXECUTOR_SECRET"),
}

# Define what each service can call
SERVICE_PERMISSIONS = {
    "triage-agent": ["research-agent:query", "tool-executor:invoke"],
    "research-agent": ["tool-executor:invoke"],
    "tool-executor": [],  # Leaf service — calls no one
}

async def verify_service_token(
    x_service_name: str = Header(...),
    x_service_token: str = Header(...),
) -> str:
    expected_secret = SERVICE_SECRETS.get(x_service_name)
    if not expected_secret:
        raise HTTPException(status_code=401, detail="Unknown service")

    if not secrets.compare_digest(x_service_token, expected_secret):
        raise HTTPException(status_code=401, detail="Invalid service token")

    return x_service_name

def require_service_permission(action: str):
    async def checker(
        service_name: str = Depends(verify_service_token),
    ) -> str:
        allowed = SERVICE_PERMISSIONS.get(service_name, [])
        if action not in allowed:
            raise HTTPException(
                status_code=403,
                detail=f"Service {service_name} not authorized for {action}",
            )
        return service_name
    return checker

Token Exchange for Cross-Boundary Calls

When an agent needs to call an external service on behalf of a user, it should exchange its service token for a scoped token rather than forwarding the user's credentials:

async def exchange_token_for_service(
    user_token: str,
    target_service: str,
    required_scopes: list[str],
) -> str:
    """Exchange a user token for a scoped service token."""
    # Validate the original user token
    user_payload = decode_token(user_token)

    # Create a constrained token for the target service
    service_payload = {
        "sub": user_payload["sub"],
        "org_id": user_payload["org_id"],
        "acting_as": "triage-agent",  # Which service is making the call
        "target": target_service,
        "scopes": required_scopes,
        "exp": datetime.now(timezone.utc) + timedelta(minutes=5),
    }
    return jwt.encode(service_payload, SECRET_KEY, algorithm=ALGORITHM)

Making Authenticated Service Calls

Build a reusable HTTP client that automatically attaches service credentials:

import httpx

class AgentServiceClient:
    def __init__(self, service_name: str, service_token: str):
        self.service_name = service_name
        self.service_token = service_token

    async def call_service(
        self, url: str, method: str = "POST", payload: dict = None,
    ) -> dict:
        headers = {
            "X-Service-Name": self.service_name,
            "X-Service-Token": self.service_token,
            "Content-Type": "application/json",
        }
        async with httpx.AsyncClient() as client:
            response = await client.request(
                method, url, json=payload, headers=headers, timeout=30.0,
            )
            response.raise_for_status()
            return response.json()

# Usage in an agent
triage_client = AgentServiceClient(
    service_name="triage-agent",
    service_token=os.environ["TRIAGE_AGENT_SECRET"],
)

result = await triage_client.call_service(
    url="https://research-agent:8443/api/query",
    payload={"question": "What are the latest metrics?"},
)

FAQ

When should I use mTLS versus service tokens?

Use mTLS when you need strong cryptographic identity at the transport layer — especially in Kubernetes environments where service meshes like Istio can automate certificate management. Use service tokens when you need application-layer authorization decisions (which service can call which endpoint with what scopes). In production multi-agent systems, using both provides defense in depth.

How do I rotate service tokens without downtime?

Support dual-token validation during rotation. When rotating, generate a new secret, update the service registry to accept both old and new secrets, deploy the new secret to the calling service, then remove the old secret. This creates an overlap window where both tokens are valid. Automate this with a secrets manager like HashiCorp Vault.

How do service meshes like Istio simplify mTLS?

Istio injects a sidecar proxy (Envoy) alongside each service pod. The sidecar handles TLS termination and certificate rotation automatically — your application code does not need to manage certificates at all. Istio's CA issues short-lived certificates (24 hours by default) and rotates them transparently. You get mTLS for free with zero application code changes.

---

#MTLS #ServiceMesh #MultiAgent #FastAPI #ZeroTrust #Microservices #AgenticAI #LearnAI #AIEngineering