Skip to main content

Join the Waitlist

Be part of a new era of securing AI agents. Limited early access spots available. Get priority access to WormAI Cloud, advanced threat protection, and enterprise-grade audit capabilities.

​
Introduction

This guide shows how an enterprise can use the WORM SDK to add a security and auditing layer for MCP-based agentic applications with just 15-30 lines of code.
  • Create a custom Gateway
  • Protect MCP servers with in-server middleware
  • Configure policies at both layers (gateway and middleware)
  • Enable fine-grained chain of custody and authentication

​
SDK Overview

The SDK is composed of packages that work together: 
worm-orchestrator: Python wrapper for agent frameworks with automatic tracing, security, identity propagation, and audit.
worm-gateway: HTTP proxy with pluggable auth, entitlements/quotas, guardrails, and routing.
worm-middleware: In-server enforcement (tool allowlists, security testing, WORM audit, idempotency).
instrumentation-node: Applicable for telemetry/spans/metrics.
worm-audit-crypto: Signing utilities for WORM evidence.
To call tools from your orchestrator, use the bridge helper detailed later in the Call Tools section.

​
Architecture Overview

  • Gateway validates inputs (AuthN/AuthZ/quotas), enforces guardrails, and forwards trace context.
  • Middleware enforces tool pinning, gating, idempotency for side effects, emits WORM audits with compliance tags.
  • Uploader (embedded with gateway and middleware) ships Audits to control-plane via presigned headers.
Every tool invocation results in a signed WORM envelope capturing: input/output hashes, identity summary (privacy-safe), policy decision and reasons, testing status and findings (if enabled), compliance tags, timing, and risk score. This enables audit-by-default without extra app code.

​
Agents Example (CrewAI)

Start with your multi-agent application. This example uses CrewAI for the payments demo. Replace with your preferred framework as needed. 
# crew_orchestrator.py (payments demo example)
import hashlib, json, time, requests
from crewai import Agent, Task, Crew
from crewai.tools import Tool

GW = "http://127.0.0.1:8088"
AUTH = "Bearer dev"
RUN_ID = f"run-{int(time.time()*1000)}"

def idem(run_id, node_id, tool_at_ver, args):
    payload = json.dumps(args or {}, separators=(",", ":"))
    h = hashlib.sha256(payload.encode()).hexdigest()[:16]
    return f"{run_id}:{node_id}:{tool_at_ver}:{h}"

def call_mcp(server, tool_at_ver, args, node_id):
    url = f"{GW}/mcp/{server}/{tool_at_ver}"
    headers = {
        "Content-Type": "application/json",
        "Authorization": AUTH,
        "x-idempotency-key": idem(RUN_ID, node_id, tool_at_ver, args),
    }
    r = requests.post(url, headers=headers, json=args or {})
    r.raise_for_status()
    return r.json()

charge = Tool(name="payments.charge", description="Charge in cents",
    func=lambda input: call_mcp("payments", "charge@1.0", {"amount": int(input)}, "executor"))
refund = Tool(name="payments.refund", description="Refund by chargeId",
    func=lambda input: call_mcp("payments", "refund@1.0", {"chargeId": input}, "executor"))

planner = Agent(role="Planner", goal="Plan safe payment ops.", backstory="You plan.", allow_delegation=False, verbose=True)
executor = Agent(role="Executor", goal="Execute ops.", backstory="You execute.", tools=[charge, refund], allow_delegation=False, verbose=True)

crew = Crew(agents=[planner, executor], tasks=[
    Task(description="Decide to charge 1999 then refund ch_123.", expected_output="Plan", agent=planner),
    Task(description="Execute: charge then refund.", expected_output="Results", agent=executor)
])
print(crew.kickoff())

​
Enterprise Orchestrator Wrapper

For enterprise applications, wrap your agent orchestration with the WORM security and observability layer. The worm-orchestrator package monitors and audits the entire orchestration flow, including agent decisions, task execution, planning logic, and information flow at the enterprise level before requests reach the gateway. 
from worm_orchestrator import WormOrchestrator

# Initialize WORM wrapper for your orchestration layer
worm = WormOrchestrator(
    gateway_url="goworm-gateway.goworm.ai",
    tenant="sample-tenant",
    auth_token="Bearer <your-token>",
    environment="prod"
)

# Wrap your existing orchestrator (CrewAI, LangGraph, AutoGen, custom)
with worm.trace_orchestration(user_id="user-123", roles=["admin"], run_id="run-001"):
    # Your existing crew/orchestrator code runs inside this context
    result = crew.kickoff()  # or your orchestrator's run method
    print(result)
The wrapper captures orchestration-level audit trails including agent decisions and reasoning, task assignments and execution order, planning steps and intermediate outputs, and how information flows between agents within your enterprise orchestrator. User identity and roles propagate securely through all layers (orchestrator → gateway → MCP servers), with automatic trace correlation, idempotency key generation, and structured error handling. The package is framework-agnostic and works seamlessly with CrewAI, LangGraph, AutoGen, or custom orchestrators.

​
Gateway

Programmatic HTTP proxy for MCP: AuthZ, quotas, guardrails, and trace context. The gateway forwards to named MCP servers using a simple routing map. 
import { startGatewayServer } from 'worm-sdk'

startGatewayServer({
  port: port_number,
  enforcement: { mode: 'enforce' }, // or 'observe'
  security: {
    egress_allowlist: ['https://api.example.com/*', 'http://127.0.0.1:*'], // prevent egress to any other URL
    fs_allowlist: ['./*'],
    limits: { max_payload_kb: 256, max_duration_ms: 10000 }, // secure from DOS attacks
    side_effects: { require_idempotency_key: true }
  },
  dedupe: {
    enabled: true,
    duplicatePolicy: 'replay', // or 'reject'
    ttlSec: num_seconds,
    namespace: 'sample-namespace',
  },
  targets: {
    payments: 'http://mcp-payments.internal:9091',
    orders: 'http://mcp-orders.internal:9092'
  } // Server name -> base host URL mapping
})

​
Gateway Parameters

Gateway parameters

- port: TCP port for HTTP /mcp proxy.
- enforcement.mode: 'enforce' blocks violations; 'observe' only logs.
- security.egress_allowlist: Allowed URL patterns for args.url; deny if not matched.
- security.fs_allowlist: Allowed path patterns for args.path; deny if not matched.
- security.limits.max_payload_kb: Max JSON body size.
- security.limits.max_duration_ms: Optional upstream timeout guard.
- security.side_effects.require_idempotency_key: Require idempotency for mutating tools.
- dedupe.enabled: Turn on dedupe.
- dedupe.duplicatePolicy: 'replay' returns cached result; 'reject' returns 409.
- dedupe.ttlSec: Cache window (10–30 min typical).
- dedupe.namespace: Key prefix for multi-tenant isolation.
- targets: name→base URL mapping; '/mcp/:server/:tool@ver' forwards to 'BASE/invoke/:tool@ver'.
Tune these to your risk posture. Start in observe to measure impact, then switch to enforce. Keep allowlists as specific as practical; prefer pinning tool versions.
Enforcement modes: enforce blocks violations; observe logs findings without blocking.
Guardrails: egress/FS allowlists and payload limits are applied before forwarding; violations become structured errors and audit entries.
Routing: /mcp/:server/:tool@ver maps to the configured server base and invokes /invoke/:tool@ver upstream.

​
Call Tools

Use the bridge helper to invoke MCP tools through the gateway (self-hosted or through WormAI Cloud). It standardizes idempotency, headers, tracing, and errors so your orchestrator code stays simple. 
import { callToolViaGateway } from 'worm-sdk'
// Example: Payments and Orders MCP servers
await callToolViaGateway({
  gatewayBaseUrl: 'custom-gateway-url or goworm-gateway.goworm.ai',
  server: 'payments',
  toolAtVersion: 'charge@1.0',
  args: { amount: 1999 },
  authHeader: 'Bearer dev',
  runId: 'run-123',
  nodeId: 'executor'
})

​
Attach Identity (optional)

Pass an identity option to set a privacy-safe x-auth-context header. The middleware parses this and includes user hash, roles, and safe JWT fields in the audit envelope. 
import { callToolViaGateway } from 'worm-sdk'

// Example: attach privacy-safe identity. The SDK will set x-auth-context
// using either userHash (preferred), a raw userId (hashed client-side),
// an enterprise assertion (compact JWS), or an Authorization header.
await callToolViaGateway({
  gatewayBaseUrl: 'custom-gateway-url or goworm-gateway.goworm.ai',
  server: 'payments',
  toolAtVersion: 'charge@1.0',
  args: { amount: 1999 },
  runId: 'run-123',
  nodeId: 'executor',
  identity: {
    tenant: 'sample-tenant',
    userId: 'user-123',      // or userHash: 'sha256-hex', or assertion: '<compact-jws>'
    roles: ['user']
  }
})

​
Parameters

  • gatewayBaseUrl: Base URL of your running gateway.
  • server: MCP server name as configured in the gateway target map.
  • toolAtVersion: Tool identifier with version, e.g., charge@1.0.
  • args: JSON payload forwarded to the tool handler.
  • authHeader: Authorization header (e.g., Bearer ...).
  • runId / nodeId: Correlate calls within a graph and seed idempotency.
  • idempotencySalt: Optional extra salt to vary the key if needed.
  • timeoutMs: Client-side timeout (default 60s).
  • identity: Build a safe identity header using userId/userHash/assertion/authorization with optional tenant and roles.

​
Behavior

  • Builds x-idempotency-key from runId, nodeId, tool@version, and an args hash for safe retries/deduplication.
  • Forwards Authorization, W3C traceparent/baggage, and optional x-tenant, x-graph-run-id, x-graph-node.
  • Non-2xx responses throw an Error with err.code and err.status for easy handling.

​
Audit

Auditing enables compliance, forensics, and explainability. Every tool invocation writes a signed envelope. The gateway’s uploader and middleware via presigned upload batches and ships these files to control-plane’s S3 for durable retention and indexing. Minimal audit envelope example: 
{
  "ts": "2025-09-27T10:34:15.824Z", // ISO timestamp when processing started
  "trace_id": "...",
  "span_id": "...",
  "tenant": "sample-tenant",
  "server": "payments@1.0",
  "tool": "charge@1.0",
  "status": "ok", // or "error"
  "latency_ms": ...,
  "retries": ...,
  "input_sha256": "...",
  "output_sha256": "...",
  "policy": { "decision": "allow", "retention": "...", "reason": null },
  "idempotency_key": "...",
  "actor": "sample-actor",
  "auth": { "scheme": "prehashed", "user_hash": "sha256-...", "tenant": "sample-tenant", "roles": ["user"] },
  "op": "create",
  "egress_domain": "api.example.com", // or null
  "fs_path": "path/to/file", // or null
  "compliance_tags": {},
  "previous_tool_version": null,
  "new_tool_version": null,
  "risk_score": 0,
  "sig": "...",
  ... // other fields
}
Security note:
  • Envelopes can include a privacy-safe auth summary.
  • Avoid secrets/PII in examples: no real tokens, keys, user IDs, emails, or bucket names.
  • Describe the shape only: scheme, user_hash or token_sha256, optional roles and minimal JWT fields; no raw tokens are stored.
  • Controls: signatures are verified on ingest and tenancy is enforced; deployments may omit email or use user_hash to reduce PII.
Fields:
  • ts: ISO timestamp when processing started; aligns with server span start.
  • trace_id / span_id: Trace correlation IDs to stitch gateway ↔ server traces.
  • tenant: Workspace identifier used for isolation and indexing.
  • server: MCP server name@version that handled the invocation.
  • tool: Tool@version invoked (must be pinned when policy requires).
  • status: ok or error outcome.
  • latency_ms / retries: End-to-end duration and retry count.
  • input_sha256 / output_sha256: SHA-256 of args/result for integrity without storing raw data.
  • policy.decision: allow or deny; retention class; optional reason when denied.
  • idempotency_key: Stable key for mutating ops; enables safe retries/deduplication.
  • auth: Privacy-safe identity context from x-auth-context or derived from Authorization: scheme, user_hash, optional roles and JWT claims subset (sub, iss, aud, exp, iat, email, tenant). No raw tokens are stored.
  • actor / op: Actor summary.
  • egress_domain / fs_path: Present when network/FS guardrails are exercised.
  • compliance_tags: Mapped tags from security checks and scanner findings.
  • risk_score / risk_tags: Derived risk for dashboards and alerts.
  • sig: Ed25519 signature over a canonicalized envelope for tamper evidence.

​
Audit Parameters

Audit parameters

- worm.dir: Local staging folder for Audits (default .worm).
- worm.retentionDays: Local hint for rotation/cleanup; policy controls true retention.

Gateway uploader (env):
- WORM_DIR: Path to local .worm directory.
- API_BASE_URL: Control-plane base Host URL.
- SVC_BEARER: Service token for presign/auth.
- CLIENT_NAME / CLIENT_VERSION / ENVIRONMENT: Used to derive S3 key prefix and tenant.

Middleware presign (worm-audit-store, Coming soon via WormAI Cloud and and a Control Plane Dashboard):
- presign.apiBaseUrl: Control-plane base.
- presign.authHeader: Authorization header (e.g., Bearer ...).
- presign.clientName: Tenant/workspace.
- presign.clientVersion: Client version marker.
- presign.environment: dev|staging|prod.

Optional direct S3 upload for enterprise control (not needed with presign):
- worm.bucket: s3://bucket/prefix for direct writes.
- AWS_REGION, AWS_S3_ENDPOINT, AWS_S3_FORCE_PATH_STYLE: SDK settings if used.
- Prefer presign when MCP servers should not hold cloud creds.

​
Configuration

Configure gateway behavior, enforcement modes, and audit export settings through environment variables or configuration files.

​
Gateway Configuration

  • Port and Routing: Define the gateway listening port and upstream MCP server endpoints.
  • Enforcement Mode: Toggle between enforce (block violations) and observe (log only) modes for testing and production.
  • Security Controls: Configure egress allowlists, filesystem access patterns, payload size limits, and request timeouts.
  • Idempotency: Enable idempotency key requirements for side-effect operations to prevent duplicate executions.

​
Audit Export

  • Storage Integration: Configure secure upload destinations for audit logs using presigned URLs or direct cloud storage credentials.
  • Tenancy and Versioning: Set client identifiers and version markers to organize audit data by tenant and deployment version.
  • Integrity Assurance: Audit files are signed per-line and bundled with SHA256 manifests to ensure tamper-evident storage and verifiable chain of custody.

​
Middleware

Wrap each MCP server to enforce tool allowlists, Application Security Testing, egress/FS rules, WORM signing/audit, and idempotency for side effects. Use withWormMeta to attach file path and scanning hints so code changes are detected precisely. If an x-auth-context header is present, the middleware parses it and includes a privacy-safe identity summary in the audit envelope. 
import http from 'node:http'
import { wormSecurityMiddleware, withWormMeta } from 'worm-middleware'
import { LocalDevSigner } from 'worm-audit-crypto'
import policy from './policy.json'

const PORT = goworm-gateway.goworm.ai / set auto-assigned port from WormAI Cloud

// Context configuration for the server and tool handlers
type Ctx = { args: any; idempotencyKey?: string|null; headers?: Record<string,string> }
const tools = new Map<string, (ctx: Ctx) => Promise<any> | any>()
const tool = (nameAtVersion: string, handler: (ctx: Ctx) => any) => tools.set(nameAtVersion, handler)
const serverLike = { name: 'payments', version: '1.0.0', tool } as any

tool('charge@1.0', withWormMeta(async ({ args }) => ({ ok: true, chargeId: 'ch_' + Date.now(), amount: args?.amount }), { filePath: __filename }))
tool('refund@1.0', withWormMeta(async ({ args }) => ({ ok: true, refundId: 'rf_' + Date.now(), chargeId: args?.chargeId }), { filePath: __filename }))

wormSecurityMiddleware({
  policy: policy as any,
  signer: LocalDevSigner.fromKeyRef((policy as any).signing.key_ref),
  otel: { endpoint: (policy as any).observability.otlp_endpoint, serviceName: 'payments', tenant: 'sample-tenant' or tenant from x-auth-context through worm gateway},
})(serverLike)

http.createServer(async (req, res) => {
  // minimal /invoke/:tool@ver HTTP adapter.
}).listen(PORT, () => console.log('[payments] :' + PORT))

​
Policy

severity_level: "high", "medium", "low", "info"
{
  "tenant": "sample-tenant",
  "signing": { "enabled": true, "key_ref": "goworm-dev.pem" },
  "observability": { "otlp_endpoint": "goworm-otlp.goworm.ai" },
  "tools": {
    "allowlist": [
      "mcp://payments/charge@1.0",
      "mcp://payments/refund@1.0",
      "mcp://orders/getOrder@1.0"
    ],
    "deny_if_unpinned_version": true
  },
  "enforcement": { "deny_on": ["tool_not_in_allowlist", "version_unpinned", "missing_idempotency_for_side_effect"] },
  "security": {
    "egress_allowlist": ["https://api.example.com/*", "http://127.0.0.1:*"] ,
    "fs_allowlist": ["./*"],
    "limits": { "max_payload_kb": 256 } // optional to prevent DOS attacks
  },
  "sast": { "enabled": true, "block_on_change": true, block_on_severity: severity_level}, 
  "dast": { "enabled": true, "block_on_change": false, "block_on_severity": severity_level}
}
Key fields:
  • tools.allowlist: Only listed mcp://server/tool@version are permitted; enable deny_if_unpinned_version to force pinning.
  • enforcement.deny_on: Reasons that become hard denies (e.g., not in allowlist, version unpinned, missing idempotency for side effects).
  • security: Egress/FS allowlists and payload limits applied inside the server process.
  • sast/dast: Toggle and tune scanning; block_on_change and block_on_severity govern gating behavior.
  • observability: OTLP endpoint and sampling for spans/metrics.

​
Authentication

Choose one: OIDC (JWT), service tokens (shared secret), or place the gateway behind your enterprise proxy and disable built-in auth. In all cases, identity drives entitlements, quotas, and audit tenancy.

​
OIDC (JWT)

import { startGatewayServer } from 'worm-sdk'
// Authenticate enterprise callers using OIDC (JWT) at the gateway
startGatewayServer({
  port: port_number,
  authn: {
    mode: 'oidc',
    oidc: {
      issuer: 'https://your-tenant.us.auth0.com/',
      jwks_uri: 'https://your-tenant.us.auth0.com/.well-known/jwks.json',
      // optional parameters
      tenantClaim: 'org',   // where tenant/org is stored in JWT
      userClaim: 'email',  
      rolesClaim: 'roles'
    }
  },
  enforcement: { mode: 'enforce' }, // or 'observe'
  security: { egress_allowlist: ['https://api.example.com/*'] },
  targets: { payments: 'http://127.0.0.1:9091', orders: 'http://127.0.0.1:9092' } // Server name -> base host URL mapping
})

​
Service Token

import { startGatewayServer } from 'worm-sdk'
// Authenticate using a shared service token (HMAC) you control
startGatewayServer({
  port: port_number,
  authn: { mode: 'service_token', serviceSecretB64: process.env.SERVICE_SECRET_B64! },
  enforcement: { mode: 'enforce' }, // or 'observe'
  targets: { payments: 'http://127.0.0.1:9091', orders: 'http://127.0.0.1:9092' } // Server name -> base host URL mapping
})

// How to mint a service token (Node.js)
import crypto from 'node:crypto'

const secretB64 = process.env.SERVICE_SECRET_B64! // base64 bytes shared with gateway
const payload = { tenant: 'acme', user: 'svc@acme', roles: ['svc'], exp: Math.floor(Date.now()/1000) + 3600 }
const payloadB64 = Buffer.from(JSON.stringify(payload)).toString('base64url')
const mac = crypto.createHmac('sha256', Buffer.from(secretB64, 'base64')).update(payloadB64).digest('base64url')
const token = `${payloadB64}.${mac}`

​
Trusted Proxy

import { startGatewayServer } from 'worm-sdk'
// Trust upstream (IAP/Zero-Trust proxy) and disable built-in auth
// Place the gateway behind your proxy; restrict network path to proxy only
startGatewayServer({
  port: port_number,
  authn: { mode: 'disabled' },
  enforcement: { mode: 'enforce' }, // or 'observe'
  // Optionally add entitlements keyed by tenant headers your proxy injects
  targets: { payments: 'http://127.0.0.1:9091', orders: 'http://127.0.0.1:9092' }
})
Recommendation: use OIDC where possible. Service tokens are suitable for service-to-service calls; prefer short expirations and rotate secrets. If trusting an upstream proxy, restrict network paths to only allow traffic from that proxy and rely on its authz headers.

​
Telemetry

Set the middleware policy’s OTLP endpoint to emit spans. The gateway forwards W3C trace context so server spans chain correctly. Configure your collector (e.g., Tempo) at the given endpoint. 
{
  "observability": {
    "otlp_endpoint": "goworm-otlp.goworm.ai", // or your own OTLP endpoint to collect traces for Grafana, Tempo, etc.
    "sample": { "success": 0.1, "error": 1 }
  }
}
  • Trace correlation: Orchestrator → Gateway → MCP server spans appear as a single trace with consistent attributes (tenant, decision, tool).
  • Sampling: Configure per-policy sampling to reduce noise in success-heavy paths while keeping 100% for errors.

​
Coming Soon

​
Control Plane Dashboard

The Control Plane Dashboard will provide easy visualization of posture metrics, policy versioning, denials, and vulnerabilities with built-in observability. Track policy changes over time, identify blocked invocations and their reasons, surface security findings from Security testing scans, and monitor compliance posture and risk scores across your entire fleet—all in a unified, real-time view.

​
WormAI Cloud

WormAI Cloud will offer additional security utilities and configurable in-server parameters that guarantee threat protection and fine-grained chain of custody.
  • Enhanced Security Utilities: Pre-built integrations for advanced threat detection, anomaly scoring, and automated incident response.
  • Configurable In-Server Parameters: Granular control over policy enforcement, retention rules, automated version management, and compliance mappings without redeploying services.
Stay tuned for early access and beta announcements.

​
About WormAI, Inc.

For partnership or inquiries, contact: Documentation and all contents © 2025 WormAI, Inc. All rights reserved. No reproduction, reverse engineering, or use for AI/ML training without written permission.