Service-to-Service Authentication for Multi-Agent Systems: mTLS and Service Tokens
Implement service-to-service authentication for multi-agent architectures using mTLS and service tokens. Covers certificate setup, trust boundaries, token exchange, and FastAPI integration.
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.
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:
See AI Voice Agents Handle Real Calls
Book a free demo or calculate how much you can save with AI voice automation.
# 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
CallSphere Team
Expert insights on AI voice agents and customer communication automation.
Try CallSphere AI Voice Agents
See how AI voice agents work for your industry. Live demo available -- no signup required.