Enterprise Security with MCP & watsonx Orchestrate: Passing User Context and Bearer Tokens to Sidecars

WXO Labs Series Icon By Markus van Kempen | Series: WXO Labs & Tutorial Guide (Part 8 of 15)
Interactive Lab Portal: https://markusvankempen.github.io/wxo-labs/
GitHub Repository: https://github.com/markusvankempen/wxo-labs

Enterprise Security with MCP & watsonx Orchestrate: Passing User Context and Bearer Tokens to Sidecars

Why This Pattern Matters

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.

πŸ“‹ Prerequisites & Setup

Ensure your environment has the required dependencies:

pip install ibm-watsonx-orchestrate

Verify active environment:

orchestrate env list

πŸ› οΈ Step 1: Implement the Logic (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)

πŸ“¦ Step 2: Declare Dependencies (requirements.txt)

starlette>=0.30.0
uvicorn>=0.25.0

πŸ€– Step 3: Configure the AI Agent (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

πŸš€ Step 4: Import & Test Live in Terminal

Import tool and test conversationally:

curl -H "Authorization: Bearer " http://localhost:8000/check-identity

Live Response:

{
  "identity_source": "WxO Connection HTTP Injection",
  "roles": ["admin"],
  "status": "authenticated",
  "user_email": "markus@ca.ibm.com"
}

πŸ’‘ Pro Tips & Gotchas

🏷️ Recommended Community Tags

When publishing this blog post, select or enter the following tags in the community editor:

watsonx Orchestrate, MCP Security, Bearer Tokens, JWT, OAuth, Enterprise Security, Starlette

πŸ—ΊοΈ Series Roadmap (15 Hands-On Labs)

This tutorial is Part 8 of 15 in the complete enterprise hands-on series:

Explore the interactive curriculum and code repository: