Verify That an Email Came From an AI Agent

import httpx
import base64
import json

MM_API_KEY = "$MULTIMAIL_API_KEY"
EMAIL_ID = "em_01HZ8QXYZ1234567890"

response = httpx.get(
    f"https://api.multimail.dev/v1/emails/{EMAIL_ID}",
    headers={"Authorization": f"Bearer {MM_API_KEY}"},
)
response.raise_for_status()
email = response.json()

identity_header = email["headers"].get("X-MultiMail-Identity")
if not identity_header:
    raise ValueError("No X-MultiMail-Identity header found — message was not sent via MultiMail agent")

payload_b64, signature_b64 = identity_header.split(".")
payload = json.loads(base64.urlsafe_b64decode(payload_b64 + "=="))

print("Agent ID:", payload["agent_id"])
print("Model:", payload["model_id"], "(", payload["model_provider"], ")")
print("Oversight mode:", payload["oversight_mode"])
print("AI disclosure:", payload["ai_disclosure"])
print("Sent at:", payload["sent_at"])

import httpx
import base64
import json
from cryptography.hazmat.primitives.asymmetric.ec import ECDSA
from cryptography.hazmat.primitives.hashes import SHA256
from cryptography.hazmat.primitives.serialization import load_der_public_key
from cryptography.hazmat.backends import default_backend
from cryptography.exceptions import InvalidSignature
from jwt.algorithms import ECAlgorithm

MM_API_KEY = "$MULTIMAIL_API_KEY"

def verify_identity_header(identity_header: str, tenant_id: str) -> dict:
    payload_b64, signature_b64 = identity_header.split(".")

    "cm"># Fetch tenant public key (JWK format)
    jwk_response = httpx.get(
        f"https://api.multimail.dev/v1/identity/public-key",
        params={"tenant_id": tenant_id},
        headers={"Authorization": f"Bearer {MM_API_KEY}"},
    )
    jwk_response.raise_for_status()
    jwk = jwk_response.json()

    "cm"># Load public key from JWK
    public_key = ECAlgorithm.from_jwk(json.dumps(jwk))

    "cm"># Verify signature over the raw payload bytes
    signed_message = payload_b64.encode("ascii")
    signature = base64.urlsafe_b64decode(signature_b64 + "==")

    try:
        public_key.verify(signature, signed_message, ECDSA(SHA256()))
    except InvalidSignature:
        raise ValueError("X-MultiMail-Identity signature is INVALID — message may be tampered or spoofed")

    payload = json.loads(base64.urlsafe_b64decode(payload_b64 + "=="))
    return payload


"cm"># Usage
payload = verify_identity_header(identity_header, tenant_id=payload["tenant_id"])
print("Signature VALID")
print(json.dumps(payload, indent=2))

import { createPublicKey, verify } from 'crypto';

const MM_API_KEY = process.env.MM_API_KEY!;
const BASE_URL = 'https://api.multimail.dev';

interface IdentityPayload {
  agent_id: string;
  model_id: string;
  model_provider: string;
  tenant_id: string;
  oversight_mode: 'read_only' | 'gated_all' | 'gated_send' | 'monitored' | 'autonomous';
  ai_disclosure: boolean;
  mailbox: string;
  sent_at: string;
  nonce: string;
}

async function verifyAgentEmail(emailId: string): Promise<{ valid: boolean; payload: IdentityPayload }> {
  const emailRes = await fetch(`${BASE_URL}/v1/emails/${emailId}`, {
    headers: { Authorization: `Bearer ${MM_API_KEY}` },
  });
  if (!emailRes.ok) throw new Error(`Failed to fetch email: ${emailRes.status}`);
  const email = await emailRes.json();

  const identityHeader: string | undefined = email.headers?.['X-MultiMail-Identity'];
  if (!identityHeader) throw new Error('Missing X-MultiMail-Identity header');

  const [payloadB64, signatureB64] = identityHeader.split('.');
  const payload: IdentityPayload = JSON.parse(
    Buffer.from(payloadB64, 'base64url').toString('utf8')
  );

  "cm">// Fetch public key for this tenant
  const keyRes = await fetch(
    `${BASE_URL}/v1/identity/public-key?tenant_id=${payload.tenant_id}`,
    { headers: { Authorization: `Bearer ${MM_API_KEY}` } }
  );
  if (!keyRes.ok) throw new Error(`Failed to fetch public key: ${keyRes.status}`);
  const jwk = await keyRes.json();

  const publicKey = createPublicKey({ key: jwk, format: 'jwk' });
  const signatureBytes = Buffer.from(signatureB64, 'base64url');
  const signedMessage = Buffer.from(payloadB64, 'ascii');

  const valid = verify('sha256', signedMessage, publicKey, signatureBytes);
  if (!valid) throw new Error('Signature verification failed');

  "cm">// Policy assertions
  if (!payload.ai_disclosure) {
    throw new Error(`EU AI Act violation: ai_disclosure is false for agent ${payload.agent_id}`);
  }

  const sentAt = new Date(payload.sent_at).getTime();
  const ageMs = Date.now() - sentAt;
  if (ageMs > 5 * 60 * 1000) {
    console.warn(`Identity header is ${Math.round(ageMs / 1000)}s old — check for replay`);
  }

  return { valid: true, payload };
}

"cm">// Audit log
const { valid, payload } = await verifyAgentEmail('em_01HZ8QXYZ1234567890');
console.log(JSON.stringify({
  event: 'identity_verification',
  result: valid ? 'PASS' : 'FAIL',
  agent_id: payload.agent_id,
  model_id: payload.model_id,
  oversight_mode: payload.oversight_mode,
  ai_disclosure: payload.ai_disclosure,
  verified_at: new Date().toISOString(),
}));

// In Claude Desktop or any MCP client connected to MultiMail:

// Step 1: Read the email and inspect headers
read_email({
  email_id: "em_01HZ8QXYZ1234567890"
})
// Returns: { subject, from, body, headers: { "X-MultiMail-Identity": "...", ... }, agent_metadata: { agent_id, model_id, oversight_mode, ai_disclosure } }

// Step 2: If the email is in the pending/approval queue, decide on it after verifying identity
decide_email({
  email_id: "em_01HZ8QXYZ1234567890",
  decision: "approve",
  reason: "Identity header verified — agent_id matches expected automation, ai_disclosure confirmed"
})

// The MCP server surfaces agent_metadata as a first-class field so you do not need
// to manually decode the X-MultiMail-Identity header in conversational workflows.
// For programmatic verification pipelines, use the REST API approach above.

"cm"># GET /v1/emails/{email_id}/identity-receipt returns a signed JSON receipt
"cm"># containing the identity payload, the verification public key fingerprint,
"cm"># and a MultiMail-issued timestamp. Safe to store in audit logs as-is.

curl -s \
  -H "Authorization: Bearer $MM_API_KEY" \
  "https://api.multimail.dev/v1/emails/em_01HZ8QXYZ1234567890/identity-receipt" \
  | jq &"cm">#039;{
      agent_id: .payload.agent_id,
      model_id: .payload.model_id,
      oversight_mode: .payload.oversight_mode,
      ai_disclosure: .payload.ai_disclosure,
      key_fingerprint: .key_fingerprint,
      receipt_issued_at: .issued_at
    }&"cm">#039;

"cm"># Example output:
"cm"># {
"cm">#   "agent_id": "agt_support_responder_v2",
"cm">#   "model_id": "claude-sonnet-4-6",
"cm">#   "oversight_mode": "monitored",
"cm">#   "ai_disclosure": true,
"cm">#   "key_fingerprint": "SHA256:k3Fg9...",
"cm">#   "receipt_issued_at": "2026-04-19T14:22:31Z"
"cm"># }