attest-mcp

@attest-dev/sdk/mcp

适用于 Model Context Protocol 服务的 Attest 凭证强制执行中间件。

只需两行代码，即可保护现有 MCP 服务中的每次工具调用。

安装

npm install @attest-dev/sdk

@attest-dev/sdk 将核心客户端和 MCP 中间件打包在同一个软件包中，并提供两个入口点。

两行代码完成集成

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { withAttest } from "@attest-dev/sdk/mcp";

const server = new McpServer({ name: "my-tools", version: "1.0.0" });
const protectedServer = withAttest(server, {
  issuerUri: "https://api.attest.dev",
});

// Register tools exactly as before — every call is now credential-gated.
protectedServer.tool("send_email", schema, handler);

withAttest 会就地修改 server.tool()，并返回与原始类型相同的服务对象。其他所有功能——connect()、close() 和传输方式——均保持不变。

工作原理

每次工具调用时，都会在处理程序执行前：

1. 提取凭证：从 extra.authInfo.token（MCP 身份验证中间件路径）、extra.meta.attest_token 或 extra.meta.authorization 中提取
2. 验证 RS256 签名和过期时间：离线对照发行方的 JWKS 验证（默认仅获取一次，并缓存 1 小时）
3. 将工具名称映射到作用域：使用 scopeForTool() 映射，并检查凭证的 att_scope 是否涵盖该作用域
4. 检查撤销状态：通过 GET /v1/revoked/{jti} 检查（进程内缓存 10 秒）
5. 继续执行：成功时调用原始处理程序
6. 返回结构化错误：任何检查失败时，Agent 框架都会收到包含类型化 JSON 负载且 isError: true 的响应
7. 发送审计事件：向 POST /v1/audit 发送事件——异步执行，不会阻塞工具调用

作用域映射

工具名称会自动映射为 resource:action 作用域字符串。第一个词（第一个 _ 之前的部分）作为操作，其余部分作为资源。

工具名称	所需作用域	备注
send_email	email:send
read_file	file:read
list_documents	documents:list → documents:read	list 映射为 read
create_user	user:create → user:write	create 映射为 write
delete_record	record:delete
get_weather_forecast	weather_forecast:read	多词资源会保留下划线
run_query	query:run → query:execute	run 映射为 execute

操作别名——以下原始前缀会映射为规范操作：

原始前缀	规范操作
create、update、put、patch、set、upsert	write
get、fetch、list、search、query、find、lookup	read
remove、destroy	delete
run、invoke、call	execute

多词资源会在作用域字符串中保留其下划线。get_weather_forecast → weather_forecast:read。这是有意为之——完整名词短语会作为资源保留。签发凭证时，请使用相同字符串：scope: ["weather_forecast:read"]。

在注册时覆盖——将 AttestToolOptions 对象作为 server.tool() 的最后一个参数传入。对于非标准工具名称，这是设置作用域的首选方式，因为作用域紧邻工具定义：

import { withAttest } from "@attest-dev/sdk/mcp";
import type { AttestToolOptions } from "@attest-dev/sdk/mcp";

protectedServer.tool("gh_create_issue", schema, handler, {
  requiredScope: "github:write",
} satisfies AttestToolOptions);

protectedServer.tool("stripe_charge", schema, handler, {
  requiredScope: "stripe:write",
} satisfies AttestToolOptions);

// Auto-mapping still applies when requiredScope is omitted.
protectedServer.tool("send_email", schema, handler);
// → "email:send"

withAttest 会使用该选项对象，并在 MCP SDK 收到参数前将其移除，因此 SDK 不会感知到它。

全局覆盖：在选项中使用 toolScopeMap（适用于所有未显式声明 requiredScope 的工具）：

const protectedServer = withAttest(server, {
  issuerUri: "https://api.attest.dev",
  toolScopeMap: {
    "send_message": "slack:send",
    "query_db":     "postgres:read",
  },
});

作用域发现端点

getAttestScopes() 会返回实时作用域注册表——每个工具名称及其解析后的作用域字符串。可将其接入 HTTP 端点，以便凭证签发方查询服务所需的作用域，无需带外协调。

import { withAttest, getAttestScopes } from "@attest-dev/sdk/mcp";

const protectedServer = withAttest(server, { issuerUri: "https://api.attest.dev" });

protectedServer.tool("send_email", schema, handler);
protectedServer.tool("gh_create_issue", schema, handler, { requiredScope: "github:write" });

// Express / Hono / any HTTP framework:
app.get("/.well-known/attest-scopes", (_req, res) => {
  res.json({ tools: getAttestScopes(protectedServer) });
});

响应：

{
  "tools": {
    "send_email":      "email:send",
    "gh_create_issue": "github:write"
  }
}

编排器 Agent 会先获取此端点，再签发凭证：

// Before starting a task, discover what the MCP server needs:
const { tools } = await fetch("https://my-mcp-server/.well-known/attest-scopes")
  .then(r => r.json());

// Issue a credential with exactly those scopes:
const { token } = await attestClient.issue({
  agent_id: "my-agent",
  user_id:  "usr_alice",
  scope:    Object.values(tools),   // ["email:send", "github:write"]
  instruction: "...",
});

无需维护作用域字符串电子表格。无需在服务端和客户端之间手动同步。服务端就是事实来源。

被阻止时 Agent 会收到的内容

{
  "content": [
    {
      "type": "text",
      "text": "{\n  \"error\": \"attest_violation\",\n  \"reason\": \"scope_violation\",\n  \"detail\": \"Credential (jti: abc-123) grants [research:read] but tool \\\"send_email\\\" requires \\\"email:send\\\".\",\n  \"jti\": \"abc-123\",\n  \"taskId\": \"tid-xyz\"\n}"
    }
  ],
  "isError": true
}

isError: true 表示 MCP 框架这是一条错误响应。reason 字段可供机器读取：

reason	含义
no_credential	请求中未找到 Attest JWT
invalid_credential	JWT 签名验证失败或格式错误
credential_expired	JWT 的 exp 已过期
credential_revoked	JTI 在撤销列表中
scope_violation	凭证未授予所需的作用域

选项

withAttest(server, {
  // Required: URI of your Attest server
  issuerUri: "https://api.attest.dev",

  // How long to cache JWKS before re-fetching (seconds). Default: 3600
  jwksCacheTTL: 3600,

  // How long to cache revocation status per JTI (seconds).
  // Set 0 to disable caching (every call hits the server). Default: 10
  revocationCacheTTL: 10,

  // false = log violations via onViolation but don't block the call.
  // Useful for gradual rollout. Default: true
  requireCredential: true,

  // Per-tool scope overrides. Keys are exact tool names.
  toolScopeMap: {
    "send_message": "slack:send",
  },

  // Called on every scope violation (synchronous).
  // Use for metrics, alerting, structured logging.
  onViolation: (event) => {
    console.warn("Attest violation", {
      tool:     event.toolName,
      reason:   event.reason,
      required: event.requiredScope,
      granted:  event.grantedScope,
      jti:      event.jti,
    });
  },
});

在工具处理程序中读取凭证声明

withAttest 验证凭证后，使用 getAttestContext 读取声明，无需再次验证：

import { withAttest, getAttestContext } from "@attest-dev/sdk/mcp";

protectedServer.tool("send_email", schema, async (args, extra) => {
  const ctx = getAttestContext(extra);
  // ctx.att_uid   — the human who initiated the task
  // ctx.att_tid   — task tree ID (use for audit correlation)
  // ctx.att_scope — granted scopes
  // ctx.att_depth — delegation depth (0 = root)
  // ctx.jti       — this credential's unique ID

  await sendEmail(args.to, args.body, { actingAs: ctx?.att_uid });
  return { content: [{ type: "text", text: "sent" }] };
});

逐步推出

将 Attest 添加到现有服务中，在确认映射正确之前，不会阻止流量：

const protectedServer = withAttest(server, {
  issuerUri: "https://api.attest.dev",
  requireCredential: false,   // observe, don't block
  onViolation: (event) => metrics.increment("attest.violation", {
    tool:   event.toolName,
    reason: event.reason,
  }),
});

监控中的违规率降至零后，将 requireCredential 切换为 true。

审计日志

每次工具调用——无论成功还是失败——都会向你的 Attest 服务发送 POST /v1/audit。投递异步进行，绝不会阻塞工具执行。

审计投递尽力而为，并以结构化方式报告错误。若要确保投递成功，请配置 onAuditError 并实现自己的重试队列。

失败会按优先级依次报告，绝不会被静默丢弃：

1. onAuditError(error, event) — 由你处理（重试队列、备用日志、寻呼通知）
2. onViolation({ reason: "audit_failure", ... }) — 未设置 onAuditError 时，会报告到现有监控系统
3. console.warn — 未配置任何回调时的最后手段

const protectedServer = withAttest(server, {
  issuerUri: "https://api.attest.dev",

  // Tier 1: implement a retry queue for guaranteed delivery
  onAuditError: (err, event) => {
    retryQueue.push({ event, attempts: 0, nextRetry: Date.now() + 5000 });
    logger.error("attest audit delivery failed", {
      jti:      event.jti,
      agent_id: event.agent_id,
      error:    err.message,
    });
  },

  // Tier 2: violations (including audit_failure) go to your monitoring
  onViolation: (e) => metrics.increment("attest.event", { reason: e.reason }),
});

审计端点返回非 2xx 响应时，会被视为投递失败，而非静默成功。

审计事件包含：event_type、jti、att_tid、att_uid、agent_id（mcp:<toolName>）、scope，以及 outcome 元字段（allowed、scope_violation、blocked_revoked）。

查询任务树的审计链：

import { AttestClient } from "@attest-dev/sdk";

const client = new AttestClient({ baseUrl: "https://api.attest.dev", apiKey: "..." });
const chain = await client.audit(taskId);
chain.events.forEach(e => console.log(e.event_type, e.jti, e.created_at));

完整示例

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { withAttest } from "@attest-dev/sdk/mcp";
import { z } from "zod";

const server = new McpServer({ name: "email-server", version: "1.0.0" });

server.tool(
  "send_email",
  { to: z.string().email(), subject: z.string(), body: z.string() },
  async ({ to, subject, body }) => {
    await sendEmail(to, subject, body);
    return { content: [{ type: "text", text: `Sent to ${to}` }] };
  },
);

// Wrap — every tool now requires a valid credential with "email:send" scope.
const protectedServer = withAttest(server, {
  issuerUri: process.env.ATTEST_URI ?? "http://localhost:8080",
  onViolation: (e) => console.warn("blocked:", e.reason, e.toolName),
});

const transport = new StdioServerTransport();
await protectedServer.connect(transport);

许可证

Apache-2.0