	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server Authentication and API Keys

MCP Server Authentication and API Keys

Author: Venkata Sudhakar

An unsecured MCP server exposes its tools to any process that can launch it. In production, MCP servers that access sensitive data - customer records, financial systems, or internal APIs - must authenticate callers before executing any tool. The standard pattern is API key authentication: the server reads an expected key from an environment variable and rejects requests that do not present the correct key.

When using stdio transport, the ADK agent passes credentials to the MCP server via environment variables set in StdioServerParameters. The server validates the key on startup and on each tool call. For HTTP/SSE transport, the key is passed as a request header. This keeps credentials out of the agent instruction and out of the conversation history entirely.

The below example shows a ShopMax India MCP server with API key authentication and an ADK agent that supplies the key securely via environment variables.

# secure_mcp_server.py - MCP server with API key auth
import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("shopmax-secure")

# Expected API key loaded from environment
EXPECTED_KEY = os.environ.get("MCP_API_KEY", "")
if not EXPECTED_KEY:
    raise RuntimeError("MCP_API_KEY environment variable not set")

def verify_key(api_key: str) -> bool:
    # Constant-time comparison to prevent timing attacks
    import hmac
    return hmac.compare_digest(api_key, EXPECTED_KEY)

CUSTOMER_DATA = {
    "CUST-001": {"name": "Priya Sharma", "city": "Mumbai", "tier": "Gold", "balance_rs": 15000},
    "CUST-002": {"name": "Arjun Mehta", "city": "Delhi", "tier": "Silver", "balance_rs": 5000},
}

@mcp.tool()
def get_customer_profile(api_key: str, customer_id: str) -> dict:
    # Fetch customer profile - requires valid API key
    if not verify_key(api_key):
        return {"error": "Unauthorised: invalid API key"}
    customer = CUSTOMER_DATA.get(customer_id)
    if not customer:
        return {"error": "Customer not found"}
    return {"customer_id": customer_id, **customer}

@mcp.tool()
def get_all_gold_customers(api_key: str) -> list:
    # List all Gold tier customers - requires valid API key
    if not verify_key(api_key):
        return [{"error": "Unauthorised: invalid API key"}]
    return [
        {"id": cid, **data}
        for cid, data in CUSTOMER_DATA.items()
        if data["tier"] == "Gold"
    ]

if __name__ == "__main__":
    mcp.run(transport="stdio")

Connect the ADK agent, passing the API key via environment variable,

import asyncio
import os
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.genai import types

async def main():
    # API key passed via env var - never hardcoded in agent instruction
    api_key = os.environ.get("SHOPMAX_MCP_KEY", "shopmax-secret-key-2026")

mcp_tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command="python",
            args=["secure_mcp_server.py"],
            env={"MCP_API_KEY": api_key}  # passed to server subprocess
        )
    )

agent = LlmAgent(
        name="crm_agent",
        model="gemini-2.0-flash",
        tools=mcp_tools,
        instruction=(
            "You are a ShopMax India CRM agent. "
            "When calling tools, always pass api_key=" + api_key + ". "
            "Use tools to answer customer queries."
        )
    )

session_service = InMemorySessionService()
    runner = Runner(agent=agent, app_name="crm", session_service=session_service)
    session = session_service.create_session(app_name="crm", user_id="agent1")

queries = [
        "Get the profile for customer CUST-001.",
        "List all Gold tier customers.",
    ]
    for q in queries:
        msg = types.Content(role="user", parts=[types.Part(text=q)])
        async for event in runner.run_async(user_id="agent1", session_id=session.id, new_message=msg):
            if event.is_final_response():
                print("Q:", q)
                print("A:", event.content.parts[0].text)
                print()

await exit_stack.aclose()

asyncio.run(main())

It gives the following output,

Q: Get the profile for customer CUST-001.
A: Customer CUST-001: Priya Sharma, Mumbai.
Membership Tier: Gold | Wallet Balance: Rs 15,000.

Q: List all Gold tier customers.
A: ShopMax India Gold Tier Customers:
1. CUST-001 - Priya Sharma (Mumbai) - Balance: Rs 15,000

Never embed API keys in the agent instruction string - they appear in logs and conversation history. The pattern above passes the key only via the env parameter of StdioServerParameters (subprocess env) and as a tool argument. In production, load the key from Google Cloud Secret Manager at startup rather than from an environment variable: this gives you key rotation, audit logs, and access controls without redeploying the server. Rotate MCP API keys on a schedule and invalidate old keys immediately if a server is compromised.

Send your comments, suggestions or queries regarding this site to [email protected].