Add Auth to Your MCP Server

This guide walks through adding MCP Auth to an existing MCP server. The integration is 5 lines of code in any language.

Overview

The integration has three steps:

1. Load your root public key - The server operator's identity that anchors all delegation chains
2. Extract the proof from tool arguments
3. Verify the proof against your root key

The library handles signature verification, delegation chain walking, caveat checking, and argument cleaning.

TypeScript

Install

npm install @kanoniv/agent-auth

Integration

import {
  verifyMcpToolCall,
  identityFromBytes,
  hexToBytes,
  type McpAuthMode,
} from "@kanoniv/agent-auth";

// Load root public key from environment (hex-encoded, 64 chars)
const rootPubKeyHex = process.env.KANONIV_ROOT_PUBLIC_KEY!;
const rootIdentity = identityFromBytes(hexToBytes(rootPubKeyHex));

// Choose auth mode from environment
const authMode = (process.env.KANONIV_MCP_AUTH ?? "optional") as McpAuthMode;

// In your tool handler:
function handleToolCall(toolName: string, args: Record<string, unknown>) {
  const outcome = verifyMcpToolCall(toolName, args, rootIdentity, authMode);

  if (outcome.verified) {
    console.log(`Agent: ${outcome.verified.invoker_did}`);
    console.log(`Root: ${outcome.verified.root_did}`);
    console.log(`Chain depth: ${outcome.verified.depth}`);
  }

  // outcome.args has _proof stripped - pass to your tool logic
  return executeToolLogic(toolName, outcome.args);
}

Complete MCP server example

A minimal MCP server with auth verification on every tool call:

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import {
  verifyMcpToolCall,
  identityFromBytes,
  hexToBytes,
  type McpAuthMode,
} from "@kanoniv/agent-auth";

// --- Configuration ---
const rootPubKeyHex = process.env.KANONIV_ROOT_PUBLIC_KEY!;
const rootIdentity = identityFromBytes(hexToBytes(rootPubKeyHex));
const authMode = (process.env.KANONIV_MCP_AUTH ?? "optional") as McpAuthMode;

// --- Server setup ---
const server = new Server({ name: "my-mcp-server", version: "1.0.0" }, {
  capabilities: { tools: {} },
});

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "resolve",
      description: "Resolve an entity by source and external ID",
      inputSchema: {
        type: "object",
        properties: {
          source: { type: "string" },
          external_id: { type: "string" },
        },
        required: ["source", "external_id"],
      },
    },
  ],
}));

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  // --- Auth verification (5 lines) ---
  let outcome;
  try {
    outcome = verifyMcpToolCall(name, args ?? {}, rootIdentity, authMode);
  } catch (err) {
    return {
      content: [{ type: "text", text: `Auth failed: ${err}` }],
      isError: true,
    };
  }

  // outcome.args is clean (no _proof), outcome.verified has identity info
  const agent = outcome.verified?.invoker_did ?? "anonymous";

  // --- Tool logic ---
  if (name === "resolve") {
    const { source, external_id } = outcome.args as {
      source: string;
      external_id: string;
    };
    return {
      content: [{
        type: "text",
        text: `Resolved ${source}:${external_id} (caller: ${agent})`,
      }],
    };
  }

  return {
    content: [{ type: "text", text: `Unknown tool: ${name}` }],
    isError: true,
  };
});

// --- Start ---
const transport = new StdioServerTransport();
await server.connect(transport);

Handling failures

verifyMcpToolCall throws a CryptoError when verification fails. The error has a code field for programmatic handling:

import { CryptoError } from "@kanoniv/agent-auth";

try {
  const outcome = verifyMcpToolCall(name, args, rootIdentity, "required");
} catch (err) {
  if (err instanceof CryptoError) {
    switch (err.code) {
      case "SIGNATURE_INVALID":
        // Bad signature, broken chain, missing proof, or DID mismatch
        return { content: [{ type: "text", text: "Unauthorized" }], isError: true };
      case "INVALID_KEY_LENGTH":
        // Malformed public key in proof
        return { content: [{ type: "text", text: "Bad proof format" }], isError: true };
      default:
        return { content: [{ type: "text", text: `Auth error: ${err.message}` }], isError: true };
    }
  }
  throw err;
}

Rust

Install

cargo add kanoniv-agent-auth

Integration

use kanoniv_agent_auth::mcp::{verify_mcp_tool_call, McpAuthMode};
use kanoniv_agent_auth::AgentIdentity;

// Load root public key from environment (hex-encoded, 64 chars)
let root_hex = std::env::var("KANONIV_ROOT_PUBLIC_KEY").unwrap();
let root_bytes = hex::decode(&root_hex).unwrap();
let root_identity = AgentIdentity::from_bytes(&root_bytes).unwrap();

// Parse auth mode from environment
let mode = McpAuthMode::from_str_lossy(
    &std::env::var("KANONIV_MCP_AUTH").unwrap_or("optional".into())
);

// In your tool handler:
fn handle_tool_call(
    tool_name: &str,
    args: &serde_json::Value,
    root_identity: &AgentIdentity,
    mode: McpAuthMode,
) -> Result<serde_json::Value, String> {
    let outcome = verify_mcp_tool_call(tool_name, args, root_identity, mode)
        .map_err(|e| format!("Auth failed: {}", e))?;

    if let Some(ref verified) = outcome.verified {
        println!("Agent: {} (depth: {})", verified.invoker_did, verified.depth);
    }

    // outcome.args has _proof stripped
    execute_tool_logic(tool_name, &outcome.args)
}

Manual extract + verify

If you want more control over the verification flow:

use kanoniv_agent_auth::mcp::{McpProof, verify_mcp_call};
use kanoniv_agent_auth::AgentIdentity;

fn handle_tool_call(args: &serde_json::Value, root: &AgentIdentity) {
    let (proof, clean_args) = McpProof::extract(args);

    if let Some(proof) = proof {
        match verify_mcp_call(&proof, root) {
            Ok(result) => {
                println!("Verified: {} (chain: {:?})", result.invoker_did, result.chain);
            }
            Err(e) => {
                eprintln!("Verification failed: {}", e);
                return;
            }
        }
    }

    // Use clean_args for tool logic
}

Python

Install

pip install kanoniv-agent-auth

Integration

import json
import os
from kanoniv_agent_auth import (
    AgentIdentity,
    McpProof,
    verify_mcp_call,
    extract_mcp_proof,
)

# Load root public key from environment (hex-encoded, 64 chars)
root_hex = os.environ["KANONIV_ROOT_PUBLIC_KEY"]
root_identity = AgentIdentity.from_bytes(bytes.fromhex(root_hex))

def handle_tool_call(tool_name: str, args_json: str):
    # Extract proof from arguments
    proof, clean_args_json = extract_mcp_proof(args_json)

    if proof is not None:
        # verify_mcp_call returns (invoker_did, root_did, chain, depth)
        invoker_did, root_did, chain, depth = verify_mcp_call(proof, root_identity)
        print(f"Agent: {invoker_did} (depth: {depth})")

    # clean_args_json has _proof stripped
    clean_args = json.loads(clean_args_json)
    return execute_tool_logic(tool_name, clean_args)

Creating proofs (agent side)

from kanoniv_agent_auth import AgentKeyPair, Delegation, McpProof, inject_mcp_proof
import json

agent = AgentKeyPair.generate()
root = AgentKeyPair.generate()

# Get a delegation from the root
delegation = Delegation.create_root(
    root,
    agent.identity().did,
    json.dumps([{"type": "action_scope", "value": ["resolve"]}]),
)

# Create and inject proof
proof = McpProof.create(
    agent,
    "resolve",
    json.dumps({"source": "crm", "external_id": "123"}),
    delegation,
)

# inject_mcp_proof returns JSON string with _proof field added
args_with_proof = inject_mcp_proof(
    proof,
    json.dumps({"source": "crm", "external_id": "123"}),
)

Key points

• _proof is always stripped. Whether verification succeeds, fails, or is skipped, the _proof field is removed from the arguments before they reach your tool handler. It is a reserved protocol field.
• Invalid proofs are errors, not pass-throughs. If a _proof field is present but contains an invalid proof, verification fails even in optional mode. A present but malformed proof is never silently ignored.
• No external lookups. The proof contains the full delegation chain and all public keys. Verification is a pure computation - no network calls, no database queries.
• Thread-safe. All verification functions are stateless and safe to call from multiple threads concurrently.