AgentAdmit for MCP Server Operators

Last updated: May 29, 2026

Who This Is For

You're building an MCP server, or you already have one running. Maybe it's a file system bridge, a database query tool, a calendar integration, or something custom. If your MCP server is accessible to more than one AI agent, or you plan to make it public, this guide is for you.

Why Your MCP Server Needs Agent Auth

Here's the uncomfortable truth about MCP servers today: if an AI agent knows your server's URL (or stdio path), it can call any tool you expose with zero verification. For a local dev tool that only you use, that's fine. For a public MCP server used by multiple agents from multiple users, that's catastrophic.

Without auth, you have no idea which agent is calling your tools. You can't enforce access boundaries ("this agent can query, but not delete"). You can't revoke a rogue agent's access. You can't bill per-agent. And you certainly can't tell a user "here's everything your AI agent did on your behalf."

With AgentAdmit, every agent gets a scoped token the USER explicitly generated and handed to it. Every verification call goes through AgentAdmit's hosted service. There is no self-hosted validation mode. No local JWT verification. No bypass. This is mandatory introspection: every agent action is authenticated, scoped, logged, and revocable from a single point. The token proves that a human approved this agent's access to your tools, and specifically which tools. You validate it on every call. You know who's calling. You can revoke anytime.

The integration point is slightly different from a standard HTTP app. Instead of wrapping HTTP routes, you hook into your MCP tool dispatch logic. This guide shows exactly where.

This Isn't Theoretical

MCP server auth isn't a future problem. It's breaking right now.

In January 2026, three CVEs were discovered in Anthropic's own reference Git MCP server — the one developers copy as their starting template. Path traversal vulnerabilities in the code the ecosystem treats as the gold standard.

A security review of dozens of open-source MCP servers found the same three failures appearing repeatedly:

1. No caller authentication on HTTP servers. Any process that can reach the server can call any tool. The MCP spec doesn't mandate caller authentication — it's optional.
2. API keys hardcoded in source code. Found in real, published MCP servers. Committed to GitHub. In git history forever.
3. Over-privileged tool credentials. Every tool gets the same level of access. A read-only tool and a delete tool share the same credentials. If one tool is compromised via prompt injection, the attacker has full access.

These aren't edge cases. They're the default state of most MCP servers being built today.

The Gap in the MCP Ecosystem

The MCP specification defines OAuth 2.1 as the standard authorization mechanism. That handles authentication — proving who is connecting.

But authentication and authorization are not the same thing. Knowing WHO is connecting doesn't answer:

• Which specific tools should this agent be allowed to call?
• Did the user (not an admin) choose those permissions?
• Can the user revoke this specific agent's access without affecting other agents?
• Is there an audit trail of every tool call this agent made?
• Can different agents from the same user have different scopes?

The largest AI infrastructure providers have acknowledged this gap. Architecture posts from major agent platforms explicitly delegate per-user authorization to MCP server operators. Industry security researchers have called it "the hardest part" of agent deployment.

No existing solution in the MCP ecosystem provides user-mediated, per-agent, user-selected scoped authorization. AgentAdmit does.

The MCP-Specific Integration Point

In a standard HTTP app, AgentAdmit lives in middleware:

# Standard HTTP app: middleware wraps route handler
@app.get("/api/data")
@agentadmit.require_scope("read:data")
def get_data():
    ...

In an MCP server, there are no HTTP routes to wrap (or there's an HTTP transport layer you may not control). Instead, you intercept tool calls at the dispatch level, before the actual tool logic runs:

# MCP server: validate before dispatching the tool
async def handle_tool_call(name: str, arguments: dict) -> Any:
    token = extract_token(name, arguments)     # pull from args or metadata
    ctx = await validate_agentadmit_token(token)  # call AgentAdmit's verify API
    check_scope(ctx, required_scope_for(name))    # enforce per-tool scope
    return await dispatch_tool(name, arguments, ctx)  # call the actual tool

Token delivery differs by transport:

Both approaches work. STDIO requires the agent to include the token as a tool argument. HTTP lets you use standard header extraction, and lets you use the AgentAdmit Python or Node SDK directly if you're running FastAPI or Express under the hood.

Step-by-Step Integration

Step 1: Get Your API Key

Sign up at agentadmit.com. You'll get:

Store them as environment variables:

AGENTADMIT_APP_ID=app_7kBx9mQ2
AGENTADMIT_API_KEY=aa_test_Rz4pN8...

Test keys work identically to live keys: full validation, real introspection. Swap to live keys (aa_live_...) when you're ready to go to production.

That also means MCP server operators can use test keys for real staging and end-to-end agent testing before launch. QA agents, coding agents, and operator-run regression checks can validate scoped tool access against the real hosted infrastructure without exposing broad production credentials.

Step 2: Install the SDK

Python

pip install agentadmit

MCP server operators typically use FastAPI under the hood (for HTTP transport) or raw Python (for STDIO). The AgentAdmit Python SDK covers both:

import os
import requests

AGENTADMIT_APP_ID = os.environ["AGENTADMIT_APP_ID"]
AGENTADMIT_API_KEY = os.environ["AGENTADMIT_API_KEY"]
AGENTADMIT_VERIFY_URL = "https://api.agentadmit.com/v1/verify"

For STDIO-transport servers, you'll call the verification endpoint directly (shown in detail in the examples below). For HTTP-transport servers running on FastAPI, you can use the full AgentAdmitMiddleware from the SDK.

Node.js

npm install @agentadmit/sdk

const { validateAgentToken, requireScope } = require('@agentadmit/sdk');

const config = {
  appId: process.env.AGENTADMIT_APP_ID,
  apiKey: process.env.AGENTADMIT_API_KEY,
  verifyUrl: 'https://api.agentadmit.com/v1/verify',
};

Step 3: Hook Into Your JSON-RPC Message Handler

This is the core integration point. You intercept tool calls and validate the token before the tool runs.

Python (STDIO transport pattern):

def handle_tool_call(name: str, arguments: dict) -> dict:
    """
    Central dispatch for all tool calls.
    Validate the AgentAdmit token before running any tool.
    """
    # Extract token from tool arguments
    token = arguments.pop("agentadmit_token", None)
    if not token:
        raise PermissionError("agentadmit_token required in tool arguments")
    
    # Validate via AgentAdmit hosted service
    ctx = verify_agentadmit_token(token)
    
    # Scope enforcement (tool-specific)
    required = SCOPE_MAP.get(name)
    if required and required not in ctx["scopes"]:
        raise PermissionError(
            f"Missing scope '{required}' for tool '{name}'. "
            f"Granted scopes: {ctx['scopes']}"
        )
    
    # Dispatch to the tool
    return TOOL_HANDLERS[name](arguments, ctx)


def verify_agentadmit_token(token: str) -> dict:
    """
    Call AgentAdmit's hosted verification endpoint.
    Returns: {"scopes": [...], "user_id": "...", "connection_id": "..."}
    """
    response = requests.post(
        AGENTADMIT_VERIFY_URL,
        headers={
            "Authorization": f"Bearer {AGENTADMIT_API_KEY}",
        },
        timeout=5,
    )
    if response.status_code == 401:
        raise PermissionError("Invalid or expired AgentAdmit token")
    if response.status_code != 200:
        raise RuntimeError(f"AgentAdmit verification failed: {response.status_code}")
    return response.json()

Node.js (JSON-RPC handler pattern):

const { validateAgentToken } = require('@agentadmit/sdk');

async function handleToolCall(name, args) {
  // Extract token from tool arguments
  const token = args.agentadmit_token;
  delete args.agentadmit_token;
  
  if (!token) {
    throw new Error('agentadmit_token required in tool arguments');
  }
  
  // Validate via AgentAdmit hosted service
  const ctx = await validateAgentToken(token);
  
  // Scope enforcement
  const required = SCOPE_MAP[name];
  if (required && !ctx.scopes.includes(required)) {
    throw new Error(`Missing scope '${required}' for tool '${name}'`);
  }
  
  // Dispatch
  return TOOL_HANDLERS[name](args, ctx);
}

Step 4: Define Your Scopes

Scopes are the access boundaries you enforce. Define one scope per distinct capability. A good rule of thumb: one scope per tool, or per risk level (read vs. write vs. delete).

Scope naming convention for MCP tools:

tools:<tool_name>      # simple, direct mapping
tools:read_<resource>  # read-only variant
tools:write_<resource> # write/mutate variant
db:read                # higher-level grouping for database tools
db:write
db:delete
files:read
files:write

Register your scopes in agentadmit.yaml (Python SDK):

app_name: "My MCP Server"
app_id: "app_7kBx9mQ2"
api_base_url: "mcp://my-server"

scopes:
  - name: tools:read_file
    description: "Read file contents"
    category: "File System"
    
  - name: tools:write_file
    description: "Write or create files"
    category: "File System"
    
  - name: tools:list_directory
    description: "List directory contents"
    category: "File System"

In your tool dispatch, map tool names to required scopes:

SCOPE_MAP = {
    "read_file":       "tools:read_file",
    "write_file":      "tools:write_file",
    "list_directory":  "tools:list_directory",
    "query":           "db:read",
    "insert":          "db:write",
    "delete":          "db:delete",
}

Step 5: Test It

Test the verify endpoint directly:

# Exchange a test connection token (your agent would do this)
curl -X POST https://api.agentadmit.com/v1/exchange \
  -H "Authorization: Bearer aa_test_YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"connection_token": "ag_ct_YOUR_TEST_TOKEN"}'

# Response includes ag_at_ access token
# Now test verification:
curl -X POST https://api.agentadmit.com/v1/verify \
  -H "Authorization: Bearer ag_at_YOUR_ACCESS_TOKEN" \
  -H "X-App-Id: app_7kBx9mQ2" \
  -H "X-Api-Key: aa_test_..."

Expected response:

{
  "valid": true,
  "scopes": ["tools:read_file", "tools:list_directory"],
  "user_id": "user_abc123",
  "connection_id": "conn_xyz789",
  "agent_label": "My AI Assistant",
  "expires_at": "2026-03-22T23:00:00Z"
}

Test scope rejection:

Call a tool that requires tools:write_file with a token that only has tools:read_file. The verify call succeeds (valid token), but your scope check should return a 403-equivalent error. Verify the error message tells the agent exactly which scope is missing.

STDIO Transport vs HTTP Transport

These are the two deployment patterns for MCP servers. The auth approach differs for each.

STDIO Transport

Most MCP servers run over STDIO, a subprocess pipe with no network layer. There's no HTTP, no headers. The only way to pass a token is inside the JSON-RPC message itself.

How the token travels:

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/data/report.txt",
      "agentadmit_token": "ag_at_eyJ..."
    }
  }
}

The agent must include agentadmit_token in the arguments for every tool call. You pop it before passing arguments to the actual tool handler.

MCP spec note: The STDIO transport relies on environment credentials rather than a protocol-level auth specification. Neither STDIO nor HTTP transport provides per-user, per-tool scoped access that users control. That's why AgentAdmit's argument-based pattern fills this gap directly.

Important: Verify the token on every tool call, not just once at initialization. This is mandatory introspection — every call must be validated through AgentAdmit's hosted service. This ensures revocations take effect immediately, audit logs capture every action, and usage metering is accurate.

HTTP / Streamable HTTP Transport

If your MCP server runs over HTTP (the newer Streamable HTTP transport), you get standard HTTP headers. Use the Authorization header exactly as you would in any HTTP API:

Authorization: Bearer ag_at_eyJ...

For FastAPI-based MCP servers, you can use the full AgentAdmit Python SDK middleware:

from fastapi import FastAPI, Depends
from agentadmit import AgentAdmitMiddleware
from agentadmit.auth import require_scope

app = FastAPI()
app.add_middleware(
    AgentAdmitMiddleware,
    config_path="agentadmit.yaml",
    get_current_user=get_current_user_dep,
    verify_user_token=verify_user_jwt,
)

@app.post("/mcp/tools/call")
async def handle_tool_call(
    request: ToolCallRequest,
    agent_ctx=Depends(require_scope("tools:read_file")),
):
    # agent_ctx has: user, connection, scopes
    return run_tool(request.name, request.arguments)

For Express-based MCP servers:

const { requireScope } = require('@agentadmit/sdk');

app.post('/mcp/tools/call', 
  requireScope('tools:read_file'),
  async (req, res) => {
    const agentCtx = req.agentAdmit; // injected by middleware
    const result = await runTool(req.body.name, req.body.arguments);
    res.json(result);
  }
);

HTTP transport is simpler from an auth perspective. Standard middleware just works. STDIO requires the argument-based pattern above.

Scope Design for MCP Tools

Scope design is a product decision, not just a technical one. Here are the patterns that work well for MCP servers.

One Scope Per Tool (Maximum Granularity)

Best for: servers where tools have very different risk profiles.

SCOPE_MAP = {
    "read_file":      "tools:read_file",
    "write_file":     "tools:write_file",
    "delete_file":    "tools:delete_file",
    "list_directory": "tools:list_directory",
    "execute_command": "tools:execute",      # high-risk — separate scope
}

Users see exactly what they're granting. "Write file" is distinct from "execute command." If an agent gets compromised, the blast radius is limited to whatever scopes the user granted.

Risk-Level Grouping

Best for: servers with many tools that cluster by danger level.

SCOPE_MAP = {
    # Read-only tools — low risk
    "read_file":      "files:read",
    "list_directory": "files:read",
    "search_files":   "files:read",
    
    # Mutating tools — medium risk
    "write_file":     "files:write",
    "create_dir":     "files:write",
    "rename_file":    "files:write",
    
    # Destructive tools — high risk
    "delete_file":    "files:delete",
    "delete_dir":     "files:delete",
}

Users pick a risk tier. Simpler mental model but less granular.

Naming Conventions

Follow the namespace:action pattern. Be consistent:

# Good — clear namespace, clear action
tools:read_file
tools:write_file
db:read
db:write
db:delete
calendar:read_events
calendar:create_event
calendar:delete_event

# Avoid — too vague
tools:access
db:use
calendar:all

The scope names your users see when they generate tokens. Make them human-readable. "Delete records from the database" is better than db:del but db:delete is fine if your UI labels explain it.

What NOT to Scope

Don't create a single tools:all scope and call it done. That defeats the purpose. Users lose the ability to grant minimum-necessary access. If your MCP server's most dangerous tool is execute_command or delete_records, it must have its own scope so users can choose to exclude it.

Example: A File System MCP Server (Full Working Code)

This is a minimal MCP file server. It exposes three tools: read_file, write_file, list_directory. We show the before (vulnerable) and after (protected with AgentAdmit).

BEFORE: No Auth (Vulnerable)

import json
import sys
import os

TOOLS = {
    "read_file": {
        "description": "Read file contents",
        "inputSchema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"]
        }
    },
    "write_file": {
        "description": "Write content to a file",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    },
    "list_directory": {
        "description": "List directory contents",
        "inputSchema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"]
        }
    },
}

def handle_message(message: dict) -> dict:
    method = message.get("method")
    params = message.get("params", {})
    msg_id = message.get("id")

    if method == "initialize":
        return {"jsonrpc": "2.0", "id": msg_id, "result": {
            "protocolVersion": "2024-11-05",
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "file-server", "version": "1.0"}
        }}

    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": msg_id, "result": {
            "tools": [{"name": k, **v} for k, v in TOOLS.items()]
        }}

    if method == "tools/call":
        name = params.get("name")
        args = params.get("arguments", {})
        # ANY AGENT THAT FINDS THIS SERVER CAN RUN ANY TOOL
        # No verification. No scopes. No audit trail.
        result = run_tool(name, args)
        return {"jsonrpc": "2.0", "id": msg_id, "result": {"content": [{"type": "text", "text": result}]}}

    return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": "Method not found"}}


def run_tool(name: str, args: dict) -> str:
    if name == "read_file":
        with open(args["path"]) as f:
            return f.read()
    if name == "write_file":
        with open(args["path"], "w") as f:
            f.write(args["content"])
        return f"Written to {args['path']}"
    if name == "list_directory":
        return "\n".join(os.listdir(args["path"]))
    raise ValueError(f"Unknown tool: {name}")


def main():
    for line in sys.stdin:
        message = json.loads(line.strip())
        response = handle_message(message)
        sys.stdout.write(json.dumps(response) + "\n")
        sys.stdout.flush()

if __name__ == "__main__":
    main()

This server is completely open. Any agent that knows its path can read, write, or list any file. No questions asked.

AFTER: Protected with AgentAdmit

import json
import sys
import os
import requests

# --- AgentAdmit config ---
AGENTADMIT_APP_ID = os.environ["AGENTADMIT_APP_ID"]
AGENTADMIT_API_KEY = os.environ["AGENTADMIT_API_KEY"]
AGENTADMIT_VERIFY_URL = "https://api.agentadmit.com/v1/verify"

# --- Scope map: tool name → required scope ---
SCOPE_MAP = {
    "read_file":      "tools:read_file",
    "write_file":     "tools:write_file",
    "list_directory": "tools:list_directory",
}

# --- Tool schemas (note: agentadmit_token added to each tool's inputSchema) ---
TOOLS = {
    "read_file": {
        "description": "Read file contents",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "agentadmit_token": {"type": "string", "description": "AgentAdmit access token"}
            },
            "required": ["path", "agentadmit_token"]
        }
    },
    "write_file": {
        "description": "Write content to a file",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"},
                "agentadmit_token": {"type": "string", "description": "AgentAdmit access token"}
            },
            "required": ["path", "content", "agentadmit_token"]
        }
    },
    "list_directory": {
        "description": "List directory contents",
        "inputSchema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "agentadmit_token": {"type": "string", "description": "AgentAdmit access token"}
            },
            "required": ["path", "agentadmit_token"]
        }
    },
}


def verify_token(token: str) -> dict:
    """
    Verify an AgentAdmit token via the hosted service.
    Returns the validated context dict on success.
    Raises PermissionError on invalid/expired tokens.
    """
    try:
        resp = requests.post(
            AGENTADMIT_VERIFY_URL,
            headers={
                "Authorization": f"Bearer {AGENTADMIT_API_KEY}",
            },
            json={"token": token},
            timeout=5,
        )
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"Cannot reach AgentAdmit: {e}")

    if resp.status_code == 401:
        raise PermissionError("Invalid or expired token")
    if resp.status_code != 200:
        raise RuntimeError(f"AgentAdmit returned {resp.status_code}")

    return resp.json()
    # Returns: {"valid": true, "scopes": [...], "user_id": "...", "connection_id": "..."}


def check_scope(ctx: dict, tool_name: str) -> None:
    """Raise PermissionError if the token doesn't have the required scope for this tool."""
    required = SCOPE_MAP.get(tool_name)
    if not required:
        return  # Tool has no scope requirement (e.g., health checks)
    
    granted = ctx.get("scopes", [])
    if required not in granted:
        raise PermissionError(
            f"Missing scope '{required}' for tool '{tool_name}'. "
            f"Granted scopes: {granted}. "
            f"The user can add this permission in their AgentAdmit settings."
        )


def handle_message(message: dict) -> dict:
    method = message.get("method")
    params = message.get("params", {})
    msg_id = message.get("id")

    if method == "initialize":
        return {"jsonrpc": "2.0", "id": msg_id, "result": {
            "protocolVersion": "2024-11-05",
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "file-server", "version": "1.0"}
        }}

    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": msg_id, "result": {
            "tools": [{"name": k, **v} for k, v in TOOLS.items()]
        }}

    if method == "tools/call":
        name = params.get("name")
        args = dict(params.get("arguments", {}))

        # 1. Extract token from arguments
        token = args.pop("agentadmit_token", None)
        if not token:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {
                "code": -32600,
                "message": "agentadmit_token required in tool arguments"
            }}

        # 2. Validate the token
        try:
            ctx = verify_token(token)
        except PermissionError as e:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {
                "code": -32600,
                "message": f"Authentication failed: {e}"
            }}
        except RuntimeError as e:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {
                "code": -32603,
                "message": f"Auth service error: {e}"
            }}

        # 3. Check scope for this specific tool
        try:
            check_scope(ctx, name)
        except PermissionError as e:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {
                "code": -32600,
                "message": str(e)
            }}

        # 4. Run the tool — now we know this agent is authorized
        try:
            result = run_tool(name, args)
        except Exception as e:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {
                "code": -32603,
                "message": f"Tool error: {e}"
            }}

        return {"jsonrpc": "2.0", "id": msg_id, "result": {
            "content": [{"type": "text", "text": result}]
        }}

    return {"jsonrpc": "2.0", "id": msg_id, "error": {
        "code": -32601,
        "message": "Method not found"
    }}


def run_tool(name: str, args: dict) -> str:
    """Actual tool implementations — clean, no auth logic here."""
    if name == "read_file":
        with open(args["path"]) as f:
            return f.read()
    if name == "write_file":
        with open(args["path"], "w") as f:
            f.write(args["content"])
        return f"Written to {args['path']}"
    if name == "list_directory":
        return "\n".join(os.listdir(args["path"]))
    raise ValueError(f"Unknown tool: {name}")


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        message = json.loads(line)
        response = handle_message(message)
        sys.stdout.write(json.dumps(response) + "\n")
        sys.stdout.flush()

if __name__ == "__main__":
    main()

What changed:

• Every tool call now requires agentadmit_token in the arguments
• Token is validated via the AgentAdmit hosted service before any tool runs
• Scope is checked per-tool: an agent with tools:read_file cannot call tools:write_file
• The token carries the user identity. You know exactly whose agent made each call
• Revoking the user's token in AgentAdmit immediately blocks the agent's next call

What the agent sends (STDIO):

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/data/report.txt",
      "agentadmit_token": "ag_at_eyJhbGciOiJSUzI1NiIsInR5..."
    }
  }
}

Example: A Database Query MCP Server (Full Working Code)

A database server is the highest-stakes MCP use case. The difference between db:read and db:delete is the difference between a useful agent and a data disaster. This example makes that line explicit.

BEFORE: No Auth (Vulnerable)

import json
import sys
import sqlite3

DB_PATH = os.environ.get("DB_PATH", "app.db")

TOOLS = {
    "query":  {"description": "Run a SELECT query", "inputSchema": {"type": "object", "properties": {"sql": {"type": "string"}}, "required": ["sql"]}},
    "insert": {"description": "Insert a row",       "inputSchema": {"type": "object", "properties": {"table": {"type": "string"}, "data": {"type": "object"}}, "required": ["table", "data"]}},
    "delete": {"description": "Delete rows",         "inputSchema": {"type": "object", "properties": {"table": {"type": "string"}, "where": {"type": "string"}}, "required": ["table", "where"]}},
}

def handle_message(message: dict) -> dict:
    method = message.get("method")
    params = message.get("params", {})
    msg_id = message.get("id")

    if method == "tools/call":
        name = params.get("name")
        args = params.get("arguments", {})
        # Any agent that knows this server path can query, insert, or delete.
        # Your entire database is exposed.
        result = run_db_tool(name, args)
        return {"jsonrpc": "2.0", "id": msg_id, "result": {"content": [{"type": "text", "text": result}]}}

    # ... initialize, tools/list handlers omitted for brevity


def run_db_tool(name: str, args: dict) -> str:
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    
    if name == "query":
        cur.execute(args["sql"])
        rows = cur.fetchall()
        return json.dumps(rows)
    
    if name == "insert":
        cols = ", ".join(args["data"].keys())
        placeholders = ", ".join("?" * len(args["data"]))
        cur.execute(f"INSERT INTO {args['table']} ({cols}) VALUES ({placeholders})", list(args["data"].values()))
        conn.commit()
        return f"Inserted 1 row into {args['table']}"
    
    if name == "delete":
        cur.execute(f"DELETE FROM {args['table']} WHERE {args['where']}")
        conn.commit()
        return f"Deleted {cur.rowcount} rows from {args['table']}"
    
    conn.close()
    raise ValueError(f"Unknown tool: {name}")

Every agent that discovers this server (by URL leak, by MCP index, by prompt injection in another tool) can DELETE FROM users WHERE 1=1.

AFTER: Protected with AgentAdmit

import json
import sys
import os
import sqlite3
import requests

# --- AgentAdmit config ---
AGENTADMIT_APP_ID = os.environ["AGENTADMIT_APP_ID"]
AGENTADMIT_API_KEY = os.environ["AGENTADMIT_API_KEY"]
AGENTADMIT_VERIFY_URL = "https://api.agentadmit.com/v1/verify"
DB_PATH = os.environ.get("DB_PATH", "app.db")

# --- Scope map: query=read, insert=write, delete=delete ---
SCOPE_MAP = {
    "query":  "db:read",
    "insert": "db:write",
    "delete": "db:delete",
}

TOOLS = {
    "query": {
        "description": "Run a SELECT query",
        "inputSchema": {
            "type": "object",
            "properties": {
                "sql":              {"type": "string", "description": "SELECT statement to execute"},
                "agentadmit_token": {"type": "string", "description": "AgentAdmit access token"}
            },
            "required": ["sql", "agentadmit_token"]
        }
    },
    "insert": {
        "description": "Insert a row",
        "inputSchema": {
            "type": "object",
            "properties": {
                "table":            {"type": "string"},
                "data":             {"type": "object"},
                "agentadmit_token": {"type": "string"}
            },
            "required": ["table", "data", "agentadmit_token"]
        }
    },
    "delete": {
        "description": "Delete rows matching a WHERE clause",
        "inputSchema": {
            "type": "object",
            "properties": {
                "table":            {"type": "string"},
                "where":            {"type": "string", "description": "SQL WHERE clause (e.g. 'id = 5')"},
                "agentadmit_token": {"type": "string"}
            },
            "required": ["table", "where", "agentadmit_token"]
        }
    },
}


def verify_token(token: str) -> dict:
    """Verify token via AgentAdmit hosted service. Raises on failure."""
    try:
        resp = requests.post(
            AGENTADMIT_VERIFY_URL,
            headers={
                "Authorization": f"Bearer {AGENTADMIT_API_KEY}",
            },
            json={"token": token},
            timeout=5,
        )
    except requests.exceptions.RequestException as e:
        raise RuntimeError(f"Cannot reach AgentAdmit: {e}")

    if resp.status_code == 401:
        raise PermissionError("Invalid or expired token")
    if resp.status_code != 200:
        raise RuntimeError(f"AgentAdmit returned {resp.status_code}")
    return resp.json()


def check_scope(ctx: dict, tool_name: str) -> None:
    required = SCOPE_MAP.get(tool_name)
    if not required:
        return
    granted = ctx.get("scopes", [])
    if required not in granted:
        raise PermissionError(
            f"Missing scope '{required}' for '{tool_name}'. "
            f"Granted: {granted}"
        )


def handle_message(message: dict) -> dict:
    method = message.get("method")
    params = message.get("params", {})
    msg_id = message.get("id")

    if method == "initialize":
        return {"jsonrpc": "2.0", "id": msg_id, "result": {
            "protocolVersion": "2024-11-05",
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "db-server", "version": "1.0"}
        }}

    if method == "tools/list":
        return {"jsonrpc": "2.0", "id": msg_id, "result": {
            "tools": [{"name": k, **v} for k, v in TOOLS.items()]
        }}

    if method == "tools/call":
        name = params.get("name")
        args = dict(params.get("arguments", {}))

        # 1. Extract and verify token
        token = args.pop("agentadmit_token", None)
        if not token:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {
                "code": -32600,
                "message": "agentadmit_token required"
            }}

        try:
            ctx = verify_token(token)
        except PermissionError as e:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32600, "message": str(e)}}
        except RuntimeError as e:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32603, "message": str(e)}}

        # 2. Enforce scope
        # An agent with only db:read cannot call delete.
        # An agent with only db:write cannot query.
        # Every grant is explicit — human chose which tools their agent gets.
        try:
            check_scope(ctx, name)
        except PermissionError as e:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32600, "message": str(e)}}

        # 3. Run the tool
        try:
            result = run_db_tool(name, args)
        except Exception as e:
            return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32603, "message": f"DB error: {e}"}}

        return {"jsonrpc": "2.0", "id": msg_id, "result": {
            "content": [{"type": "text", "text": result}]
        }}

    return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": "Method not found"}}


def run_db_tool(name: str, args: dict) -> str:
    """DB tool implementations — no auth logic here."""
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()

    try:
        if name == "query":
            # Safety: only allow SELECT statements
            sql = args["sql"].strip()
            if not sql.upper().startswith("SELECT"):
                raise ValueError("Only SELECT queries are allowed via the query tool")
            cur.execute(sql)
            rows = cur.fetchall()
            cols = [d[0] for d in cur.description] if cur.description else []
            return json.dumps({"columns": cols, "rows": rows})

        if name == "insert":
            cols = ", ".join(args["data"].keys())
            placeholders = ", ".join("?" * len(args["data"]))
            cur.execute(
                f"INSERT INTO {args['table']} ({cols}) VALUES ({placeholders})",
                list(args["data"].values())
            )
            conn.commit()
            return f"Inserted 1 row into '{args['table']}'"

        if name == "delete":
            cur.execute(f"DELETE FROM {args['table']} WHERE {args['where']}")
            conn.commit()
            return f"Deleted {cur.rowcount} rows from '{args['table']}'"

        raise ValueError(f"Unknown tool: {name}")

    finally:
        conn.close()


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        message = json.loads(line)
        response = handle_message(message)
        sys.stdout.write(json.dumps(response) + "\n")
        sys.stdout.flush()

if __name__ == "__main__":
    main()

The key difference from the before version:

BEFORE: Any agent can call delete. DELETE FROM users WHERE 1=1 is one tool call away.
AFTER:  
  - Agent must present a token the USER generated
  - Token must have 'db:delete' scope (user explicitly chose to grant delete access)
  - If the user only granted 'db:read', the agent gets a 403-equivalent error on any mutating tool
  - User can revoke the token at any time via their app. Agent loses access on the next verify call.
  - You have a full audit trail: which agent, which user, which tool, when

How the scopes work in practice for a database server:

A cautious user generates a token with only db:read. Their AI assistant can query but cannot insert or delete, even if the agent decides it needs to. A power user generates a token with all three scopes. If the agent ever behaves unexpectedly, the user revokes. The next verify call returns 401. Done.

Your Users Need a Token Generation Page

Your MCP server's users need a way to generate tokens. The AgentAdmit React SDK gives you a drop-in AgentAdmit page:

import { AgentAdmitPanel } from '@agentadmit/react';

<AgentAdmitPanel
  apiBase="/agentadmit"
  authToken={userSessionToken}
  scopeResources={yourMcpScopes}
  templates={yourTemplates}
  appName="Your MCP Server"
/>

This gives users: scope selection, token generation, "How It Works" guide, templates, and connection management. Same component every AgentAdmit app owner uses.

Your MCP server needs a web frontend where users can select scopes and generate their token. Use the AgentAdmit React SDK to add this to your site.

Which AI Agents Can Use AgentAdmit Tokens?

Any agent that can make HTTP API calls and include a token in tool arguments or headers:

• OpenClaw agents (built-in HTTP tool use)
• Claude (tool use capabilities)
• Custom agents built with OpenAI API (function calling)
• Google Gemini (function calling)
• Custom agents (LangChain, CrewAI, AutoGen, etc.)
• Automation platforms (n8n, Make)

AgentAdmit is agent-agnostic. Any agent that can make HTTP requests works.

Best Practices & Recommendations

Before going live, review these recommendations:

• Do not expose internal AI tools to external agents. If your MCP server has tools that trigger AI processing, do not make those tools agent-accessible. The user's AI agent can read the raw data and do its own analysis. See "In-App AI Tools" below.
• Implement per-user, per-tool rate limiting. AgentAdmit handles authorization. You handle volume. Use the user_id from the verify response to set limits appropriate for each tool.
• Isolate agent-accessible tools. Only register tools with AgentAdmit scopes that you explicitly want agents to use. Keep administrative and internal tools separate.
• Use test keys in staging before switching to live keys. Validate your entire integration with test keys first. The behavior is identical — only the environment changes.
• Monitor your dashboard regularly. Check active connections, alert events, and usage patterns. Set up webhook delivery for alerts so your system can react automatically.

In-App AI Tools

If your MCP server has tools that trigger internal AI processing (e.g., a tool that calls GPT-4o to analyze data), do NOT expose those as agent-accessible tools. The user's AI agent IS an AI. It can read the raw data and do the analysis itself. Exposing AI-powered tools to agents creates double cost: the user pays for their agent AND your server pays for the internal AI call. Instead, give agents read access to the raw data. Let the agent do its own analysis.

Rate Limiting for MCP Server Operators

AgentAdmit's rate limits on your verification calls

When your MCP server verifies agent tokens through AgentAdmit's hosted service, those calls are subject to rate limits based on your plan:

• Test keys: Lower rate limits appropriate for development and staging
• Starter ($50/mo): Production rate limits (250K calls/mo included)
• Builder ($100/mo): Higher production rate limits (500K calls/mo included)
• Pro ($200/mo): High volume production (1M calls/mo included)
• Enterprise: Custom limits available — contact us

When rate limited, AgentAdmit returns HTTP 429 with Retry-After, X-RateLimit-Limit, X-RateLimit-Remaining, and X-RateLimit-Reset headers. All AgentAdmit SDKs handle this automatically with exponential backoff and retry.

Protect your MCP server tools from agent overuse

AgentAdmit verifies that an agent is authorized to use specific tools with specific scopes. But it does not control how frequently the agent calls those tools.

As an MCP server operator, you should implement rate limiting on your tools to prevent:

• Agents calling expensive tools too frequently
• Runaway agents hammering your database or external APIs
• A single agent consuming disproportionate resources

Use the user_id from the AgentAdmit verify response to set per-user, per-agent rate limits on your tools:

# Example: rate limit per user per tool
from agentadmit import verify_token

result = verify_token(token)
user_id = result.user_id

# Check your rate limiter before executing the tool
if not rate_limiter.allow(key=f"{user_id}:{tool_name}", limit=30, window=60):
    return {"error": "rate_limited", "message": "Too many requests. Please wait."}

# Tool execution continues...

This gives you per-user billing visibility, per-user rate limiting, and and protection against agent abuse, all using data AgentAdmit already provides.

Pricing

MCP server operators are app owners. Same pricing.

Every tool call that touches a protected tool = one verification call. Size your plan accordingly. A high-traffic MCP server running 100K tool calls/day = ~3M/mo. Pro tier covers 1M; overages at $0.20/1K beyond that. Full pricing details at agentadmit.com/pricing.

If your usage consistently exceeds your current tier, upgrading gets you more included calls at a lower overage rate. Volume pricing available for high-usage apps. Contact us through the dashboard.

Test keys are available for development and testing. Build and test before going live. Pay when you launch with live keys.

Test keys still use AgentAdmit's hosted service. Mandatory introspection applies during testing exactly as it does in production. Terms of Service and Privacy Policy acceptance should happen before test key issuance because hosted verification, logging, and metering begin during development.

Common Questions

Q: Do I need to modify my MCP tools to accept the token argument? Yes. For STDIO transport, the token travels as a tool argument. You add agentadmit_token to each tool's inputSchema and pop it before passing arguments to your actual tool logic. For HTTP transport, the token comes in the Authorization header and you don't need to change your tool schemas.

Q: Does AgentAdmit work with the official MCP Python SDK (mcp package)? Yes. The AgentAdmit verification is just an HTTP call to our hosted service. It works alongside any MCP framework. You call verify_token() inside your tool handler, wherever that lives in your framework's dispatch pattern.

Q: Can I cache the token verification result to reduce latency? No. Every call must go through AgentAdmit's hosted verification. This is mandatory introspection — it ensures revocations take effect immediately, audit logs capture every call, and usage metering is accurate. Caching bypasses all three.

Q: What if AgentAdmit's verify endpoint is down? Your call to /v1/verify will timeout or return a 5xx. Handle it gracefully. Return an error to the agent explaining the auth service is temporarily unavailable. Don't fail open (allow the call through). Fail closed (reject it). AgentAdmit's uptime SLA covers Standard and Enterprise plans.

Q: My MCP server handles multiple users. How does AgentAdmit handle multi-tenancy? Each user generates their own token. The verify response includes user_id, so you know which user's agent is calling. You can use this to scope database queries to the correct user's data. This is standard pattern. See the ctx["user_id"] value in the verify response.

Q: What about server-to-server MCP calls (agent calling an MCP server that calls another MCP server)? The token the user granted stays with that user's agent. When that agent calls your MCP server, it presents its token. If your MCP server itself needs to call a downstream service, it uses YOUR server's credentials for that downstream call, not the user's token. AgentAdmit handles the user→agent→your-server layer. Your server→downstream layer is your architecture to manage.

Q: Can I skip adding agentadmit_token to every tool schema and just document it out-of-band? Technically yes, but don't. Adding it to the inputSchema tells the agent (and any AI agent using your tool list to understand capabilities) that a token is required. Well-behaved agents will include it automatically. If you don't put it in the schema, you'll get errors from agents that don't know to include it.

Q: Do I need to pay to start building? Build and test with test keys. Pay when you go live with live keys.

What's Next

When you've integrated AgentAdmit into your MCP server:

1. Register your scopes in the AgentAdmit dashboard so we can serve them in the discovery document. When a user generates a token for your MCP server, they'll see your scope names and descriptions, not just a raw token.

2. Instrument your tools: if you're running a high-value MCP server, use the user_id from the verify response to attribute database queries, file operations, or API calls to the specific user whose agent is calling. This gives you per-user billing, per-user rate limiting, and a real audit trail.

3. Build your user-facing token generation page: Use the AgentAdmit React SDK to add a scope selection and token generation page to your website. Your users sign in, select which tools their agent can access, choose a duration, and get their token.

4. Watch for HTTP transport: Streamable HTTP is gaining traction as a deployment pattern for production MCP servers. When you move from STDIO to HTTP transport, switch to the Authorization header pattern and consider using the full AgentAdmit SDK middleware instead of raw HTTP calls to the verify endpoint.

Running Your MCP Server as a Business

Everything above gets your server protected. This section is about running it as a business: monitoring, billing, automation, and control.

View All Active Connections

Your AgentAdmit dashboard at agentadmit.com shows:

• Every active agent connection to your server
• Which user created each connection
• What scopes each agent has (which tools it can access)
• When each connection was created and when it expires
• Connection activity (last used, total calls)

Embed the <AgentAdmitAdminPanel> React component in your own admin dashboard for full visibility: Connections (all users, search/filter, revoke), Usage (calls vs tier), Alerts (thresholds + kill switch), Activity (full audit trail). Or use the AgentAdmit dashboard at agentadmit.com.

Use the Audit Trail for Billing

Every tool call validated through AgentAdmit is logged: which user, which agent, which tool, when. This gives you per-user usage data that you can use to:

• Bill users based on how many tool calls their agents make
• Identify your most active users and most popular tools
• Set per-user rate limits based on their plan
• Generate invoices with real usage data

You don't need to build your own usage tracking. Mandatory introspection already captures it.

Give Your Own AI Agent Admin Access

You can use AgentAdmit to monitor your OWN server. Generate a token with admin scopes, give it to your personal AI agent, set the duration to "until I revoke." Want full automation? Schedule your agent to monitor your server:

• "How many users connected today?"
• "Which tools are getting the most calls?"
• "Are any agents behaving unusually?"
• "Alert me if a user exceeds 10,000 calls in a day"

This is the same user-controlled access your users get, but applied to your own operations. You set the scopes. You control the duration. Your agent works while you sleep.

You define admin-level scopes in your app's scope configuration. These are scopes you grant to your own agent but not to your users' agents.

Admin Revocation (Operator Override)

As the MCP server operator, you can revoke any user's agent connection from your AgentAdmit dashboard at agentadmit.com, or via the API:

curl -X POST https://api.agentadmit.com/v1/revoke \
  -H "Authorization: Bearer aa_live_YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"connection_id": "conn_..."}'

Use this when:

• You detect an agent abusing your tools
• A user reports their agent is compromised
• You need to cut off access during an incident

Your users can also revoke their own connections from their AgentAdmit page. But as the operator, you have the override to revoke any connection across your entire user base.

Your own AI agent can monitor for abuse patterns and alert you through the dashboard.

Full API reference: agentadmit.com/docs/api App dashboard: agentadmit.com Support: support@agentadmit.com

Security: Connection Token Storage

If your MCP server generates Connection Tokens on behalf of users (rather than delegating to the AgentAdmit hosted service), you MUST comply with the AgentAdmit spec §5.3.1:

• Hash before storing: Compute SHA-256(raw_token) and store only the hash. Never persist the plaintext.
• Show once: Return the plaintext to the user exactly once. No endpoint or UI should retrieve or redisplay a previously generated token.
• Look up by hash: When an agent presents a Connection Token for exchange, hash the presented value and match against stored hashes.

This applies to any persistent storage layer: SQL, NoSQL, Redis, or in-memory stores that are serialized to disk. All AgentAdmit SDKs use mandatory introspection — every verification call goes through the hosted service. The Python and Node.js SDKs include token generation routes (the endpoint that creates connection tokens for your users), so they handle the hash-before-store pattern automatically. The Go, Java, PHP, and Ruby SDKs don't include token generation routes (the hosted service handles that), so connection token storage doesn't apply to them.

Rationale: Connection Tokens are bearer secrets. A database breach that exposes plaintext tokens gives an attacker a window (up to 15 minutes per token) to impersonate users. Hashing eliminates this risk at negligible cost.

Security Alerts and Kill Switch

MCP server operators SHOULD configure anomaly detection alerts for their AgentAdmit integrations. The alert system monitors agent behavior through the mandatory introspection audit trail and surfaces unusual patterns.

Recommended Alert Configuration for MCP Servers

MCP servers typically handle tool calls, discrete operations that agents invoke. The following thresholds are recommended starting points for MCP server operators:

Kill Switch for MCP Servers

MCP server operators SHOULD enable the kill switch for failed scope attempts. An agent repeatedly probing for unauthorized tool access is a strong signal of compromise or misconfiguration. Recommended: auto-revoke after 10 failed scope attempts in 10 minutes.

Webhook Delivery

App owners can configure webhook delivery for alerts. When an alert fires, AgentAdmit sends an HTTP POST to the configured endpoint:

{
  "event": "agentadmit.alert",
  "alert_id": "alert_a1b2c3d4e5f67890",
  "alert_type": "failed_scope_attempts",
  "severity": "warning",
  "connection_id": "conn_...",
  "message": "Unauthorized scope probing: 5 denied requests in 10 minutes",
  "action_taken": "none",
  "data": {
    "failure_count": 5,
    "threshold": 5,
    "window_minutes": 10
  },
  "timestamp": "2026-04-20T21:30:00Z"
}

For kill switch triggers, action_taken will be "auto_revoked" and severity will be "critical".

Notifying Your Users Is Your Responsibility

AgentAdmit detects anomalies, fires alerts, and (with kill switch) auto-revokes connections. How you notify your own users about these events is up to you.

AgentAdmit provides the data. You deliver it through your own system — in-app notifications, email, push, or however your app communicates with users.

1. Poll alerts via SDK — Use list_alerts() (Python) or listAlerts() (Node) from your backend to check for new events, then notify users through your existing notification system.
2. Webhook delivery — Configure a webhook URL in your AgentAdmit dashboard. When an alert fires, AgentAdmit POSTs the payload above to your server. Your server handles notification to the affected user.
3. React SDK for user self-service — Embed the <AlertsPanel> component so users can view their own alert history and tighten their thresholds directly.

This design is intentional. You know your users, your notification infrastructure, and your UX. AgentAdmit gives you the security data — you decide how to surface it.