Every B2B SaaS company today faces the same request: “Can I connect your platform to my AI agent?”Your customers want to use Claude Desktop or Cursor to interact with their data—leads in your CRM, tickets in your HelpDesk, or infrastructure in your Cloud Dashboard.But exposing your B2B API directly to an LLM client is dangerous. You cannot ask customers to paste high-privilege API keys into third-party tools. You need a Secure Bridge.In this guide, we will build a production-ready Model Context Protocol (MCP) server for “Nexus” (our fictional B2B CRM). We will implement:
OAuth 2.1 for secure customer login (using Google as our Identity Provider).
Multi-Tenant Isolation to ensure Customer A never sees Customer B’s data.
Scoped Token Propagation to reuse your existing API permissions.
We aren’t rewriting our SaaS security. We are leveraging it. The MCP Server acts as a proxy that translates “Natural Language Intent” (from Claude) into “Scoped API Calls” (to your Backend).
It’s important to understand there are two separate OAuth flows happening:Flow 1: Claude ↔ Your MCP Server
Claude acts as the OAuth client
Your MCP server is the authorization server
Issues an “MCP Token” that Claude includes in every request
This token is just a session identifier—it has no backend permissions
Flow 2: Your MCP Server ↔ Google → Nexus Backend
Your MCP server acts as the OAuth client for Google
Google provides user identity (email, name)
Your MCP server verifies the user with your Nexus backend (/auth/verify)
Nexus backend returns a scoped API token for that user
This token has real permissions and can access customer data
The MCP Server sits in the middle, translating between these two flows. The key insight: backend tokens are fetched during MCP token creation (token exchange), not during authorization.
The secret sauce is how we handle the Scoped Token. We don’t want the user to log in for every single question. We need a way to persist their identity securely during the conversation.We will use a Encryption-at-Rest strategy within our session management.
# Phase 1: Token Exchange During Authorization Code Redemption# This happens when Claude exchanges the authorization code for an MCP tokenasync def exchange_code_for_token( self, code: str, code_verifier: str, client_id: str): # 1. Verify the OAuth 2.1 Handshake (PKCE validation) auth_code = await self.code_repo.validate_code(code, code_verifier) # 2. The auth_code already contains Google user info from the authorization flow # (email was verified when the authorization code was created) user_email = auth_code.user_email # 3. Verify user access with your Nexus backend # This is where your backend validates the user and returns a scoped API token nexus_response = await self.http_client.post( "https://api.nexus-crm.com/auth/verify", headers={ "Authorization": f"Bearer {self.settings.NEXUS_ADMIN_TOKEN}" }, json={ "email": user_email } ) if nexus_response.status_code != 200: raise ValueError("User not authorized for Nexus access") scoped_api_token = nexus_response.json().get("access_token") user_id = nexus_response.json().get("user_id") # 4. Encrypt this token before storing it in the database # We NEVER store backend tokens in plaintext encrypted_token = self.encryption.encrypt(scoped_api_token) # 5. Create the MCP Token (this is what Claude will send in future requests) mcp_token = f"mcp_{secrets.token_urlsafe(32)}" await self.token_repo.save_token( token=mcp_token, client_id=client_id, user_email=user_email, user_id=user_id, metadata={ "nexus_token_encrypted": encrypted_token, "expires_at": time.time() + 86400 # 24 hours } ) # 6. Mark the authorization code as used (one-time use only) await self.code_repo.mark_code_as_used(code) return { "access_token": mcp_token, "token_type": "Bearer", "expires_in": 86400 }
Now that we have a secure session, how do we use it?A naive developer might pass the api_token as an argument to every tool:
get_leads(api_token: str, search_term: str).Do not do this.
It leaks the token into the LLM conversation history (security risk).
It confuses the model (why does it need to know about tokens?).
It’s brittle.
Instead, we use Context Injection. The tool shouldn’t know how it’s authenticated, just that it is.
First, we define a thread-safe (and async-safe) storage for our user data. This acts as a “teleportation tunnel” from our Middleware directly to our Tools.
from contextvars import ContextVarfrom typing import Optionalfrom dataclasses import dataclass@dataclassclass UserContext: """User context data structure.""" user_id: Optional[str] = None user_email: Optional[str] = None access_token: Optional[str] = None# The "Magic Variable"# This variable is local to the specific request/thread.# It cannot be seen by other users or concurrent requests._user_context: ContextVar[Optional[UserContext]] = ContextVar( "user_context", default=None)def set_user_context( user_id: Optional[str] = None, user_email: Optional[str] = None, access_token: Optional[str] = None): """ Called by Authentication Backend at the start of EVERY request. This injects the user's credentials into a thread-safe context. """ context = UserContext( user_id=user_id, user_email=user_email, access_token=access_token ) _user_context.set(context)def get_user_context() -> Optional[UserContext]: """Get the current user context.""" return _user_context.get()def get_current_user() -> Optional[str]: """Helper to get the current user's email (useful for audit logs)""" context = get_user_context() return context.user_email if context else Nonedef clear_user_context(): """Clear the current user context after request completion.""" _user_context.set(None)
Now we write the MCP tool. Notice how clean it is. There is no authentication logic visible here. It retrieves credentials from context and makes API calls.
from fastmcp import FastMCPfrom app.context import get_user_contextimport aiohttpmcp = FastMCP("Nexus Leads")def get_headers() -> dict: """Retrieve authentication headers from user context.""" user_context = get_user_context() if not user_context or not user_context.access_token: raise ValueError("User context not initialized - authentication required") return { "Authorization": f"Bearer {user_context.access_token}", "Content-Type": "application/json" }@mcp.toolasync def search_leads(query: str, industry: str = None) -> str: """ Search for sales leads in the CRM. Returns only leads the authenticated user is allowed to see. """ # The tool retrieves auth automatically from context headers = get_headers() async with aiohttp.ClientSession() as session: async with session.get( "https://api.nexus-crm.com/leads/search", headers=headers, params={"q": query, "industry": industry} ) as response: if response.status == 401: return "Authentication failed. Please restart the conversation." if response.status == 403: return "Access denied to this resource." results = await response.json() if not results: return "No leads found matching your criteria." # Format as markdown table return format_as_table(results)def format_as_table(data: list) -> str: """Helper to format results for Claude.""" if not data: return "No results found." headers = data[0].keys() table = "| " + " | ".join(headers) + " |\n" table += "| " + " | ".join(["---"] * len(headers)) + " |\n" for row in data: table += "| " + " | ".join(str(row.get(h, "")) for h in headers) + " |\n" return table
We have moved beyond “Toy MCP Servers.”By combining Google OAuth for identity, Database-Backed Token Storage, Context Propagation for state isolation, and Scoped Tokens for access control, we’ve built a system that is:
Secure: No shared API keys. Users authenticate via their company SSO.
Private: User-scoped tokens ensure proper data isolation. Multi-layer validation prevents data leakage.
Auditable: Every API call is tied to a real user identity for compliance.
Usable: The end user just chats; the rigorous security happens transparently.
This is the architecture required to take AI Agents from Localhost Experiments → Enterprise Production.