Transport-level identity injection, JWT extraction in Starlette/FastAPI, and secure credential delegation.
When an agent queries an external database via MCP, how does the MCP server know WHO is executing the command? We cannot let the LLM see raw security tokens. In this lab, we implement transport-level HTTP Bearer token injection where WXO forwards the user's authenticated token directly in the HTTP header.
mcp_sse_server.py)from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
import base64
import json
async def check_identity(request):
"""Inspect the incoming Bearer token injected by watsonx Orchestrate."""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return JSONResponse({"error": "Unauthorized: Missing Bearer Token"}, status_code=401)
token = auth_header.split(" ")[1]
try:
payload_segment = token.split(".")[1]
decoded = base64.b64decode(payload_segment + "==").decode("utf-8")
claims = json.loads(decoded)
user_email = claims.get("email", claims.get("sub", "authenticated_user"))
roles = claims.get("roles", ["viewer"])
except Exception:
user_email = "authenticated_sso_user"
roles = ["user"]
return JSONResponse({
"status": "authenticated",
"user_email": user_email,
"roles": roles,
"identity_source": "WxO Connection HTTP Injection"
})
routes = [Route("/check-identity", endpoint=check_identity, methods=["GET"])]
app = Starlette(routes=routes)
requirements.txt)starlette>=0.30.0
uvicorn>=0.25.0
agent.yaml)spec_version: v1
kind: native
name: secure_identity_agent
title: "Secure Identity Agent"
description: "Delegates authenticated queries using connection-injected Bearer tokens."
model: ibm/granite-3-8b-instruct
tools:
- check_identity
curl -H "Authorization: Bearer " http://localhost:8000/check-identity
{
"identity_source": "WxO Connection HTTP Injection",
"roles": ["admin"],
"status": "authenticated",
"user_email": "markus@ca.ibm.com"
}
Never pass raw tokens as prompt arguments to the LLM. Always inject them at the transport layer (`Authorization: Bearer`) using WXO Connection settings.
Ensure downstream services handle token refresh or expect fresh tokens on every agent dialogue turn.