从浏览器网络请求快速生成 TypeScript API 与 MCP Server：安全实操指南

bun --version
python3 --version

mkdir -p web-api-mcp-candidate/{captures,scripts,src,test}
cd web-api-mcp-candidate
bun init -y

web-api-mcp-candidate/
├── captures/
│   ├── raw.har              # 原始文件，不提交版本库
│   ├── sanitized.har        # 清理后才允许交给模型
│   └── request.fetch.txt    # 清理后才允许交给模型
├── scripts/
│   └── sanitize_har.py
├── src/                     # 由编码代理生成候选实现
├── test/                    # 由编码代理生成测试
└── .gitignore

captures/raw.har
captures/sanitized.har
captures/request.fetch.txt
captures/*.raw.txt
.env
node_modules/

#!/usr/bin/env python3
from __future__ import annotations

import json
import sys
from pathlib import Path
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

SENSITIVE_HEADERS = {
    "authorization",
    "cookie",
    "proxy-authorization",
    "set-cookie",
    "x-api-key",
    "x-csrf-token",
    "x-xsrf-token",
}
SENSITIVE_QUERY = {
    "access_token",
    "api_key",
    "auth",
    "code",
    "key",
    "password",
    "signature",
    "token",
}


def sanitize_url(url: str) -> str:
    parts = urlsplit(url)
    query = [
        (key, "[REDACTED]" if key.lower() in SENSITIVE_QUERY else value)
        for key, value in parse_qsl(parts.query, keep_blank_values=True)
    ]
    return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(query), parts.fragment))


def sanitize_headers(headers: list[dict[str, str]]) -> list[dict[str, str]]:
    return [
        {**header, "value": "[REDACTED]"}
        if header.get("name", "").lower() in SENSITIVE_HEADERS
        else header
        for header in headers
    ]


def sanitize_har(data: dict) -> dict:
    for entry in data.get("log", {}).get("entries", []):
        request = entry.get("request", {})
        response = entry.get("response", {})

        request["url"] = sanitize_url(request.get("url", ""))
        request["headers"] = sanitize_headers(request.get("headers", []))
        request["cookies"] = []
        request["queryString"] = [
            {**item, "value": "[REDACTED]"}
            if item.get("name", "").lower() in SENSITIVE_QUERY
            else item
            for item in request.get("queryString", [])
        ]
        if "postData" in request:
            request["postData"]["text"] = "[REDACTED — review manually]"

        response["headers"] = sanitize_headers(response.get("headers", []))
        response["cookies"] = []

    return data


def self_test() -> None:
    sample = {
        "log": {
            "entries": [
                {
                    "request": {
                        "url": "https://example.test/items?token=secret&page=2",
                        "headers": [{"name": "Authorization", "value": "Bearer secret"}],
                        "cookies": [{"name": "session", "value": "secret"}],
                        "queryString": [
                            {"name": "token", "value": "secret"},
                            {"name": "page", "value": "2"},
                        ],
                        "postData": {"text": "private body"},
                    },
                    "response": {
                        "headers": [{"name": "Set-Cookie", "value": "session=secret"}],
                        "cookies": [{"name": "session", "value": "secret"}],
                    },
                }
            ]
        }
    }
    result = sanitize_har(sample)
    entry = result["log"]["entries"][0]
    rendered = json.dumps(result)
    assert "secret" not in rendered
    assert "page=2" in entry["request"]["url"]
    assert entry["request"]["postData"]["text"].startswith("[REDACTED")
    print("self-test: ok")


def main() -> None:
    if sys.argv[1:] == ["--self-test"]:
        self_test()
        return
    if len(sys.argv) != 3:
        raise SystemExit("usage: sanitize_har.py INPUT.har OUTPUT.har")

    source, destination = map(Path, sys.argv[1:])
    data = json.loads(source.read_text(encoding="utf-8"))
    cleaned = sanitize_har(data)
    destination.write_text(json.dumps(cleaned, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"wrote: {destination}")


if __name__ == "__main__":
    main()

python3 scripts/sanitize_har.py --self-test
python3 scripts/sanitize_har.py captures/raw.har captures/sanitized.har

self-test: ok
wrote: captures/sanitized.har

python3 -c 'import json; from pathlib import Path; d=json.loads(Path("captures/sanitized.har").read_text()); sensitive={"authorization","cookie","proxy-authorization","set-cookie","x-api-key","x-csrf-token","x-xsrf-token"}; parts=(part for e in d.get("log",{}).get("entries",[]) for part in (e.get("request",{}),e.get("response",{}))); assert all(not part.get("cookies") and all(h.get("value")=="[REDACTED]" for h in part.get("headers",[]) if h.get("name","").lower() in sensitive) for part in parts); print("basic secret scan: ok")'

读取 captures/sanitized.har 和 captures/request.fetch.txt，先不要写代码。

目标：生成一个最小 TypeScript API 客户端和 MCP Server 候选实现。

边界：
1. 第一版只允许只读接口；不要实现创建、修改、删除、发布、付款等操作。
2. 不得把 Cookie、Token、CSRF 值或其他凭证写入源码、测试、日志或示例。
3. 认证信息只能从环境变量或调用方注入。
4. 从抓包中识别 base URL、请求方法、参数、分页方式、成功响应和错误响应。
5. 对来源证据不足的字段标记 TODO，不要猜测。
6. MCP 工具使用明确、互不重叠的名称和输入 schema。
7. 为每个客户端方法生成 mock 测试，至少覆盖成功、空结果、401/403、429 和 5xx。
8. 所有网络调用设置超时；只对安全的只读请求做有上限的重试。

先输出：
- 接口清单；
- 认证与敏感数据风险；
- 拟生成文件树；
- 缺失信息；
- 验收计划。

我确认计划后再生成代码。

src/
├── client.ts       # HTTP 客户端、超时、错误映射
├── schemas.ts      # 输入与响应类型
├── tools.ts        # 只读 MCP 工具定义
└── server.ts       # MCP Server 入口
test/
├── client.test.ts
└── tools.test.ts

bun install
bun run typecheck
bun test

python3 -c 'from pathlib import Path; files=list(Path("src").rglob("*.ts"))+list(Path("test").rglob("*.ts")); text="\n".join(p.read_text() for p in files); assert "Bearer ey" not in text and "session=" not in text; print("source secret scan: ok")'

员工的 AI/MCP Client
        │ HTTPS + 企业访问令牌
        ▼
内网 API Gateway / SSO
  - TLS
  - OIDC/JWT 校验
  - 清除外部身份头
  - 转发原始 Bearer Token 供 MCP 二次验签
  - 限流、审计、请求大小限制
        │ 仅内网或同机回环访问
        ▼
只读 MCP Server（Streamable HTTP，/mcp）
        │ 用户身份下推 / 受限服务账号
        ▼
目标网站或企业内部 API

src/
├── client.ts
├── schemas.ts
├── tools.ts
├── server-factory.ts   # 唯一的 MCP 工具注册入口
├── server-stdio.ts     # 个人本地模式
└── server-http.ts      # 企业共享模式
Dockerfile
compose.yaml

bun add @modelcontextprotocol/server jose zod
bun add -d typescript @types/bun

import {
  createMcpHandler,
  OAuthError,
  OAuthErrorCode,
  requireBearerAuth,
  type AuthInfo,
} from "@modelcontextprotocol/server";
import { createRemoteJWKSet, jwtVerify, type JWTPayload } from "jose";
import { buildServer } from "./server-factory";

function requiredEnv(name: string): string {
  const value = process.env[name];
  if (!value) throw new Error(`missing required environment variable: ${name}`);
  return value;
}

const issuer = requiredEnv("OIDC_ISSUER");
const audience = requiredEnv("OIDC_AUDIENCE");
const jwks = createRemoteJWKSet(new URL(requiredEnv("OIDC_JWKS_URL")));
const allowedHosts = new Set(
  requiredEnv("ALLOWED_HOSTS").split(",").map((value) => value.trim()).filter(Boolean),
);
const allowedOrigins = new Set(
  requiredEnv("ALLOWED_ORIGINS").split(",").map((value) => value.trim()).filter(Boolean),
);

function scopesFrom(payload: JWTPayload): string[] {
  if (typeof payload.scope === "string") return payload.scope.split(" ").filter(Boolean);
  const scp = payload.scp;
  return Array.isArray(scp)
    ? scp.filter((value): value is string => typeof value === "string")
    : [];
}

async function verifyAccessToken(token: string): Promise<AuthInfo> {
  try {
    const { payload } = await jwtVerify(token, jwks, { issuer, audience });
    if (!payload.sub || typeof payload.exp !== "number") throw new Error("missing sub/exp");
    return {
      token,
      clientId: payload.sub,
      scopes: scopesFrom(payload),
      expiresAt: payload.exp,
    };
  } catch {
    throw new OAuthError(OAuthErrorCode.InvalidToken, "invalid or expired access token");
  }
}

const gate = requireBearerAuth({
  verifier: { verifyAccessToken },
  requiredScopes: ["mcp"],
});
const handler = createMcpHandler(
  ({ authInfo }) => buildServer(authInfo),
  { responseMode: "json", legacy: "reject" },
);

function forbidden(message: string): Response {
  return Response.json({ error: message }, { status: 403 });
}

export default {
  hostname: "0.0.0.0",
  port: Number(process.env.PORT ?? "3000"),

  async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);

    if (url.pathname === "/healthz") {
      return Response.json({ status: "ok" });
    }
    if (url.pathname !== "/mcp") {
      return new Response("not found", { status: 404 });
    }
    if (!allowedHosts.has(url.hostname)) {
      return forbidden("host not allowed");
    }

    const origin = request.headers.get("origin");
    if (origin && !allowedOrigins.has(origin)) {
      return forbidden("origin not allowed");
    }

    const auth = await gate(request);
    if (auth instanceof Response) return auth;
    return handler.fetch(request, { authInfo: auth });
  },
};

process.on("SIGTERM", async () => {
  await handler.close();
  process.exit(0);
});

import { McpServer, type AuthInfo } from "@modelcontextprotocol/server";
import { z } from "zod/v4";
import { ApiClient } from "./client";

export function buildServer(authInfo?: AuthInfo): McpServer {
  const server = new McpServer({ name: "internal-web-api", version: "1.0.0" });
  const client = new ApiClient({ callerId: authInfo?.clientId });

  server.registerTool(
    "list-items",
    {
      description: "读取当前用户有权查看的条目；最多返回 100 条",
      inputSchema: z.object({
        cursor: z.string().optional(),
        limit: z.number().int().min(1).max(100).default(20),
      }),
    },
    async ({ cursor, limit }) => {
      if (!authInfo?.scopes.includes("mcp")) {
        return {
          content: [{ type: "text", text: "insufficient_scope: requires mcp" }],
          isError: true,
        };
      }

      const result = await client.listItems({ cursor, limit });
      return {
        content: [{ type: "text", text: JSON.stringify(result) }],
      };
    },
  );

  return server;
}

ARG BUN_IMAGE=oven/bun:1
FROM ${BUN_IMAGE}

WORKDIR /app
COPY package.json bun.lock* ./
RUN bun install --frozen-lockfile --production
COPY src ./src

USER bun
EXPOSE 3000
CMD ["bun", "run", "src/server-http.ts"]

services:
  mcp:
    build:
      context: .
      args:
        BUN_IMAGE: ${BUN_IMAGE:-oven/bun:1}
    restart: unless-stopped
    read_only: true
    tmpfs:
      - /tmp:size=64m,noexec,nosuid
    cap_drop:
      - ALL
    security_opt:
      - no-new-privileges:true
    ports:
      - "127.0.0.1:3000:3000"
    environment:
      PORT: "3000"
      ALLOWED_HOSTS: "mcp.intra.example.com"
      ALLOWED_ORIGINS: "https://ai.intra.example.com"
      OIDC_ISSUER: "https://idp.intra.example.com/"
      OIDC_AUDIENCE: "https://mcp.intra.example.com/mcp"
      OIDC_JWKS_URL: "https://idp.intra.example.com/.well-known/jwks.json"
    healthcheck:
      test:
        - CMD
        - bun
        - -e
        - "fetch('http://127.0.0.1:3000/healthz').then(r => { if (r.status !== 200) process.exit(1) }).catch(() => process.exit(1))"
      interval: 30s
      timeout: 5s
      retries: 3

BUN_IMAGE=registry.intra.example.com/base/bun@sha256:REPLACE_ME \
  docker compose up -d --build
docker compose ps
curl --fail http://127.0.0.1:3000/healthz

{"status":"ok"}

export MCP_TOKEN='由企业 IdP 或开发者门户签发的短期令牌'
npx @modelcontextprotocol/inspector \
  --server-url https://mcp.intra.example.com/mcp \
  --transport http \
  --header "Authorization: Bearer $MCP_TOKEN"

npx @modelcontextprotocol/inspector --cli \
  https://mcp.intra.example.com/mcp \
  --transport http \
  --header "Authorization: Bearer $MCP_TOKEN" \
  --method tools/list

npx @modelcontextprotocol/inspector --cli \
  https://mcp.intra.example.com/mcp \
  --transport http \
  --header "Authorization: Bearer $MCP_TOKEN" \
  --method tools/call \
  --tool-name list-items \
  --tool-args-json '{"limit":5}' \
  --format json

列出我有权查看的 5 条记录，只返回名称和更新时间。

查询记录 ID demo-123；如果无权访问，原样报告权限错误，不要尝试其他账号。

告诉我当前可用的工具及其只读/写入属性，不执行任何工具。

MCP Host → stdio → 用户本机 server-stdio.ts → 用户有权访问的上游系统