案例背景
某中型 SaaS 公司（23 名工程师、800 万 ARR、产品是 B2B 营销自动化平台）从 2025 年开始就在用各种 AI 脚本处理内部工作流——自动生成测试用例、自动回复客户工单、自动跑代码评审、自动出数据报告。
到 2026 年 8 月，这些脚本已经长成了 17 个独立的 Python 服务，散布在不同工程师的笔记本和内部服务器上：
├── ai-ticket-router/ # 客户工单自动分流（3 个工程师维护）
├── ai-code-reviewer/ # 代码评审（5 个工程师维护）
├── ai-report-generator/ # 数据报告（2 个工程师维护）
├── ai-test-generator/ # 测试用例生成（3 个工程师维护）
├── ai-doc-writer/ # API 文档生成（2 个工程师维护）
├── ai-pr-description/ # PR 描述生成（1 个工程师维护）
├── ... 12 个其他零散脚本 ...
问题清单：
| 痛点 | 具体表现 | 业务影响 |
|---|---|---|
| 重复建设 | 17 个脚本里有 12 个都要调 OpenAI API、读 GitHub、写数据库 | 同样的胶水代码写了 12 遍 |
| 维护分散 | 每个脚本各自的 prompt、配置、监控、日志 | 工程师离职 = 脚本变孤儿 |
| 资源浪费 | 17 个脚本平均每天调用 8000 次 OpenAI API | 月成本 ¥28000+ |
| 没有统一管控 | prompt injection 防护、安全审计、成本控制都没有 | 一次 prompt 注入能让所有脚本同时"幻觉" |
| 无法组合 | 脚本各自独立，无法做"先代码评审再测试生成"这种串联 | 必须人工触发每个脚本 |
2026 年 9 月，OpenAI 把内部跑 Codex 的 harness 改成了对外可用的 Agents API（官方文档），原生支持：
- MCP server 注册（一个 Agent 可以挂多个 MCP）
- 多 Agent 编排（handoff / manager pattern）
- Guardrail（输入 / 输出安全过滤）
- 结构化输出（JSON Schema 强约束）
- 完整的 trace 和 token 计费
目标：用 Agents API + 自建的 6 个 MCP server，把 17 个零散脚本统一成一个可编排的 Agent 服务，每月节省 480 工时、API 成本降低 60%、所有 AI 操作纳入统一审计。
本文完整复盘 6 周的迁移过程。
工具组合与分工
| 工具 | 角色 | 选型理由 | 不可替代性 |
|---|---|---|---|
| OpenAI Agents API | 主编排引擎 | 原生支持 MCP + handoff + guardrail，比 LangGraph 简洁 60% | 高 |
| Codex CLI | 子 Agent 执行 | 处理"写代码 / 改文件 / 跑命令"类任务 | 高 |
| 自建 GitHub MCP | 代码仓库操作 | 拉代码、创建 PR、看 issue、读 workflow 文件 | 高 |
| 自建 Postgres MCP | 数据库读写 | 读写内部业务数据库（工单、报告、用户） | 高 |
| 自建 Slack MCP | 通知 + 人机协作 | 给工程师发通知、在 Slack 里收集人工审批 | 中 |
| 自建 Artifactory MCP | 工具治理 | 所有 MCP 调用统一审计、凭据托管 | 中 |
| 自建 Sentry MCP | 异常监控 | Agent 执行时的异常自动关联到 Sentry | 中 |
| Redis（自建） | Agent 状态存储 | 跨 Agent 共享上下文、handoff 中间状态 | 中 |
工具组合的核心思路：
- OpenAI Agents API 是"指挥家"——决定调用哪个子 Agent、用哪个 MCP、怎么 handoff
- 6 个 MCP 是"工具箱"——每个 MCP 是一个独立的能力单元
- Codex 是"执行者"——负责具体的代码生成、文件修改、命令执行
- Redis 是"记忆"——跨 Agent 共享状态，避免每次重新计算
整体架构图：
┌─────────────────────────────────────────────────────────────┐
│ OpenAI Agents API（主编排） │
│ │
│ ┌─────────────┐ handoff ┌─────────────┐ │
│ │ Manager │ ────────→ │ Code │ │
│ │ Agent │ │ Review Agent │ │
│ │（意图分类） │ │ │ │
│ └──────┬──────┘ └──────┬──────┘ │
│ │ handoff │ handoff │
│ ↓ ↓ │
│ ┌─────────────┐ ┌─────────────┐ │
│ │ Ticket │ │ Test │ │
│ │ Router │ │ Generator │ │
│ │ Agent │ │ Agent │ │
│ └─────────────┘ └─────────────┘ │
└──────────┬──────────────────────────────────────┬───────────┘
│ │
↓ ↓
┌─────────────────────────────────────────────────────┐
│ MCP Layer（6 个自建 MCP） │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌──────────┐ │
│ │ GitHub │ │ Postgres│ │ Slack │ │Artifactory│ │
│ │ MCP │ │ MCP │ │ MCP │ │ MCP │ │
│ └─────────┘ └─────────┘ └─────────┘ └──────────┘ │
│ ┌─────────┐ ┌─────────┐ │
│ │ Sentry │ │ Redis │ │
│ │ MCP │ │ MCP │ │
│ └─────────┘ └─────────┘ │
└─────────────────────────────────────────────────────┘
迁移策略设计
经过 2 周的设计评审，最终采用**"影子流量灰度 + 分阶段切换"**的迁移策略：
阶段 1：影子期（Week 1-2）
新 Agent 服务和 17 个旧脚本并行运行
新 Agent 服务"观察"但不实际生效
对比两者结果，收集差异数据
阶段 2：10% 流量切换（Week 3）
把 10% 的真实流量切到新 Agent 服务
人工对比新旧结果，记录问题
阶段 3：50% 流量切换（Week 4-5）
扩大到 50% 流量
修复累积问题
阶段 4：100% 切换（Week 6）
切完全部流量
下线旧脚本
进入稳态运行
为什么不用"一次性切换"：17 个脚本中有些是关键路径（客户工单分流），错误会直接影响客户体验。灰度策略把风险分摊到 6 周，每周最多承担 1/4 的风险。
实施步骤
Step 1：主 Agent 设计（Day 1-3）
主 Agent（Manager Agent）负责意图分类 + 路由分发。它不直接执行任务，而是把任务分派给合适的子 Agent。
OpenAI Agents API 的 Agent 定义（TypeScript SDK）：
import { Agent, handoff, tool } from '@openai/agents';
import { githubTools } from './mcps/github';
import { postgresTools } from './mcps/postgres';
import { slackTools } from './mcps/slack';
const managerAgent = new Agent({
name: 'Manager Agent',
instructions: `你是 Manager Agent，负责接收用户请求并分派给合适的子 Agent。
可用子 Agent：
- codeReviewAgent：代码评审（GitHub PR 触发）
- ticketRouterAgent：客户工单分流（Zendesk webhook 触发）
- reportGeneratorAgent：数据报告生成（Slack 命令触发）
- testGeneratorAgent：测试用例生成（GitHub PR 触发）
- docWriterAgent：API 文档生成（GitHub PR 触发）
分派规则：
1. 分析用户请求的意图
2. 选择最合适的子 Agent
3. 如果不确定，调用 classify_intent 工具先分类
4. 永远不要自己直接执行任务
5. 失败时 fallback 到人工（通过 Slack MCP 发通知）`,
tools: [
// 主 Agent 自己只挂意图分类工具
tool({
name: 'classify_intent',
description: '对用户请求做意图分类',
parameters: {
type: 'object',
properties: {
request: { type: 'string' },
possibleIntents: { type: 'array', items: { type: 'string' } }
},
required: ['request', 'possibleIntents']
},
execute: async ({ request, possibleIntents }) => {
// 用 embedding + cosine similarity 做意图分类
const embedding = await openai.embeddings.create({
input: request,
model: 'text-embedding-3-large'
});
// ... 省略实现
return { intent: 'code_review', confidence: 0.92 };
}
})
],
handoffs: [
handoff(codeReviewAgent),
handoff(ticketRouterAgent),
handoff(reportGeneratorAgent),
handoff(testGeneratorAgent),
handoff(docWriterAgent)
]
});
关键设计决策：
- 主 Agent 不挂 MCP——保持它的"纯路由"职责，避免它越过子 Agent 直接调用工具
- 意图分类用 embedding——比 LLM 直接分类便宜 80%、速度 10x
- 5 个固定的子 Agent——每个子 Agent 职责单一，便于独立维护
- 失败 fallback 到 Slack——Agent 搞不定的事情通过 Slack 通知人工
Step 2：子 Agent 设计（Day 4-10）
5 个子 Agent 各司其职。下面以codeReviewAgent为例：
const codeReviewAgent = new Agent({
name: 'Code Review Agent',
instructions: `你是 Code Review Agent，负责对 GitHub PR 做自动评审。
工作流程：
1. 接收 PR URL 或 PR 编号
2. 通过 GitHub MCP 拉取 PR diff、文件变更、commit 历史
3. 通过 Postgres MCP 查询代码风格规范表
4. 检查代码：
- 是否符合团队代码规范（命名、注释、复杂度）
- 是否有明显的 bug（空指针、资源泄露、SQL 注入）
- 是否有未处理的边界场景
- 是否有缺失的单元测试
5. 输出结构化评审报告（JSON Schema）
6. 通过 Slack MCP 把报告发给 PR 作者
7. 如果发现严重问题，自动 @ 安全团队
禁止：
- 不要直接修改 PR 代码（只评审，不动手）
- 不要给模糊的"看起来不错"评价
- 不要评审非本仓库的 PR`,
tools: [
// 挂 GitHub MCP 的所有工具
...githubTools,
// 挂 Postgres MCP 查询代码规范
...postgresTools,
// 挂 Slack MCP 发通知
...slackTools,
// 挂 Sentry MCP（如果评审中发现运行时报错，自动关联 issue）
...sentryTools
],
// 结构化输出
outputType: {
type: 'object',
properties: {
summary: { type: 'string' },
severity: { type: 'string', enum: ['info', 'warning', 'error', 'critical'] },
findings: {
type: 'array',
items: {
type: 'object',
properties: {
file: { type: 'string' },
line: { type: 'number' },
severity: { type: 'string' },
category: { type: 'string' },
message: { type: 'string' },
suggestion: { type: 'string' }
}
}
},
autoFixAvailable: { type: 'boolean' }
},
required: ['summary', 'severity', 'findings']
},
// Guardrail：过滤危险 PR（main 分支、release 分支、超过 5000 行的 PR）
guardrails: [
{
name: 'block-dangerous-prs',
type: 'input',
execute: async ({ input }) => {
if (input.includes('main') || input.includes('release/')) {
return { blocked: true, reason: '不允许直接评审 main 分支' };
}
if (input.match(/lines: \d{4,}/)) {
return { blocked: true, reason: '单次评审代码量超过 5000 行，需要拆分' };
}
return { blocked: false };
}
}
]
});
关键设计决策：
- 结构化输出（outputType）——评审结果强制按 JSON Schema 输出，下游 Slack 通知可以自动渲染
- Guardrail 在输入侧——直接拒绝危险 PR（main 分支、超大 PR），不浪费 token
- 每个工具只挂需要的 MCP——codeReviewAgent 不挂 Artifactory MCP（评审不需要写工件）
Step 3：MCP 集成（Day 11-25）
自建 6 个 MCP server，每个 MCP 负责一类能力。下面重点讲GitHub MCP 和 Postgres MCP 的实现（其他 MCP 类似）。
GitHub MCP（用 FastMCP 框架）：
# mcp_github/server.py
from fastmcp import FastMCP
import httpx
import os
mcp = FastMCP("GitHub MCP")
@mcp.tool()
async def get_pull_request(repo: str, pr_number: int) -> dict:
"""获取 GitHub PR 的完整信息"""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.github.com/repos/{repo}/pulls/{pr_number}",
headers={"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"}
)
return resp.json()
@mcp.tool()
async def get_pr_diff(repo: str, pr_number: int) -> str:
"""获取 PR 的 diff 内容（unified diff 格式）"""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.github.com/repos/{repo}/pulls/{pr_number}",
headers={
"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}",
"Accept": "application/vnd.github.v3.diff"
}
)
return resp.text
@mcp.tool()
async def get_pr_files(repo: str, pr_number: int) -> list:
"""列出 PR 修改的所有文件"""
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.github.com/repos/{repo}/pulls/{pr_number}/files",
headers={"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"}
)
return resp.json()
@mcp.tool()
async def post_pr_comment(repo: str, pr_number: int, body: str) -> dict:
"""在 PR 上发评论"""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments",
headers={"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"},
json={"body": body}
)
return resp.json()
@mcp.tool()
async def request_reviewers(repo: str, pr_number: int, reviewers: list[str]) -> dict:
"""请求指定工程师评审"""
async with httpx.AsyncClient() as client:
resp = await client.post(
f"https://api.github.com/repos/{repo}/pulls/{pr_number}/requested_reviewers",
headers={"Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}"},
json={"reviewers": reviewers}
)
return resp.json()
if __name__ == "__main__":
mcp.run(transport="stdio")
Postgres MCP：
# mcp_postgres/server.py
from fastmcp import FastMCP
import asyncpg
import os
mcp = FastMCP("Postgres MCP")
_pool = None
async def get_pool():
global _pool
if _pool is None:
_pool = await asyncpg.create_pool(
os.environ['DATABASE_URL'],
min_size=2,
max_size=10,
command_timeout=30
)
return _pool
@mcp.tool()
async def query(sql: str, params: list = []) -> list[dict]:
"""执行 SQL 查询（只读）"""
pool = await get_pool()
async with pool.acquire() as conn:
# 强制只读事务
await conn.execute("SET TRANSACTION READ ONLY")
rows = await conn.fetch(sql, *params)
return [dict(row) for row in rows]
@mcp.tool()
async def execute(sql: str, params: list = []) -> dict:
"""执行 SQL 写入（INSERT/UPDATE/DELETE）"""
pool = await get_pool()
async with pool.acquire() as conn:
# 强制权限校验
if not _is_safe_sql(sql):
raise ValueError(f"危险 SQL 被拒绝: {sql[:100]}")
result = await conn.execute(sql, *params)
return {"affected_rows": result}
def _is_safe_sql(sql: str) -> bool:
"""白名单校验：禁止 DROP / TRUNCATE / GRANT / 超长 UPDATE"""
sql_upper = sql.upper().strip()
forbidden = ["DROP", "TRUNCATE", "GRANT", "REVOKE", "ALTER SYSTEM"]
for kw in forbidden:
if sql_upper.startswith(kw) or f" {kw} " in sql_upper:
return False
if "UPDATE" in sql_upper and sql_upper.count("WHERE") == 0:
return False # 没有 WHERE 的 UPDATE 一律拒绝
return True
if __name__ == "__main__":
mcp.run(transport="stdio")
关键设计决策：
- 每个 MCP 是独立进程——隔离故障，codeReviewAgent 挂掉不影响 ticketRouterAgent
- 每个 MCP 有自己的凭据——通过 Artifactory MCP 集中托管
- Postgres MCP 强制只读事务 + 白名单 SQL——防止 Agent 被 prompt injection 后执行危险 SQL
Step 4：OpenAI Agents API 注册 MCP（Day 26-30）
OpenAI Agents API 原生支持 MCP 注册——每个 Agent 启动时通过 mcp_servers 参数加载：
import { MCPServerStdio } from '@openai/agents';
const githubMcp = new MCPServerStdio({
name: 'github-mcp',
command: 'python',
args: ['-m', 'mcp_github.server'],
env: {
GITHUB_TOKEN: process.env.GITHUB_TOKEN
}
});
const postgresMcp = new MCPServerStdio({
name: 'postgres-mcp',
command: 'python',
args: ['-m', 'mcp_postgres.server'],
env: {
DATABASE_URL: process.env.DATABASE_URL
}
});
const codeReviewAgent = new Agent({
name: 'Code Review Agent',
// ...
mcp_servers: [githubMcp, postgresMcp, slackMcp, sentryMcp]
// ...
});
Agent 启动时自动拉取 MCP 工具列表，无需手工注册。
Step 5：灰度切换（Day 31-Week 6）
Week 3：10% 流量切换
// 灰度路由器
function routeRequest(request: any) {
const bucket = hashUserId(request.user_id) % 100;
if (bucket < 10) {
// 10% 流量走新 Agent
return agentService.handle(request);
} else {
// 90% 流量走旧脚本
return legacyService.handle(request);
}
}
对比新旧结果（10% 流量一周收集的数据）：
| 维度 | 旧脚本 | 新 Agent |
|---|---|---|
| 平均响应时间 | 12.4 秒 | 8.1 秒 |
| 准确率 | 78% | 89% |
| Token 消耗 | 100%（基准） | 65%（更省） |
| 工程师维护工时 | 18 小时/周 | 4 小时/周 |
| 失败率 | 5.2% | 1.8% |
关键发现：新 Agent 不仅结果更好，还更省钱——因为结构化输出减少了无效 token 浪费，guardrail 减少了无效调用。
三个真实业务场景
场景 1：客户工单自动分流
原痛点：每天 800+ 客户工单，3 个工程师轮流人工分到 4 个处理团队。平均分流时间 14 分钟，客户等待时间最长 2 小时。
Agent 化后：
客户发工单（Zendesk webhook）
→ 触发 ticketRouterAgent
→ 调 GitHub MCP（拉客户账号绑定的代码仓库）
→ 调 Postgres MCP（查客户账号的订阅等级、历史工单）
→ 调 Sentry MCP（查客户最近的运行时报错）
→ 综合判断：技术工单 / 账单工单 / 功能请求 / Bug 报告
→ 分配给对应处理团队
→ 通过 Slack MCP 通知工程师
→ 全程 < 30 秒
效果：分流时间从 14 分钟降到 30 秒，客户等待时间从 2 小时降到 30 秒，工程师分流工时从 18 小时/周降到 1 小时/周。
场景 2：代码评审流水线
原痛点：23 个工程师每天提 30+ PR，平均每个 PR 评审时间 4 小时（人工）。其中 60% 的评审意见是"格式问题 / 缺少测试 / 命名不规范"这类机械性问题。
Agent 化后：
工程师提 PR（GitHub webhook）
→ 触发 codeReviewAgent
→ 调 GitHub MCP（拉 PR diff、文件变更、commit 历史）
→ 调 Postgres MCP（查代码风格规范表、查最近评审历史）
→ 调 Sentry MCP（如果代码涉及运行时报错，自动关联）
→ 输出结构化评审报告（JSON Schema）
→ 通过 GitHub MCP 在 PR 上发评审评论
→ 如果发现严重问题，调 Slack MCP @ 安全团队
→ 全程 < 45 秒
效果：评审时间从 4 小时降到 45 秒（机械性问题 100% 自动处理），人工评审只关注复杂架构问题。工程师每周节省 12 小时评审时间。
场景 3：数据报告自动化
原痛点：每周一早 9 点要出"上周业务数据报告"，PM 手工跑 SQL + 写 PPT + 发邮件，耗时 3 小时。
Agent 化后：
每周一早 8:30（cron 触发）
→ 触发 reportGeneratorAgent
→ 调 Postgres MCP（跑上周业务数据 SQL）
→ 调 Slack MCP（收集上周团队动态）
→ 调 GitHub MCP（拉上周 PR / issue / commit 统计）
→ 综合生成"上周报告"
→ 通过 Slack MCP 发到 #weekly-report 频道
→ 全程 < 2 分钟
效果：报告生成从 3 小时降到 2 分钟，PM 可以专注解读数据而不是拉数据。
踩过的 11 个坑
坑 1：Handoff 死循环
现象：managerAgent → codeReviewAgent → managerAgent → codeReviewAgent ...
原因：子 Agent 在某些情况下判定"应该让 Manager Agent 处理"，又 handoff 回去。
修复：在主 Agent 里加白名单，明确"哪些场景不要 handoff 回来"。
坑 2：Guardrail 误杀
现象：某些合法 PR（涉及 main 分支的安全修复）被 guardrail 直接拒绝。
修复：guardrail 加白名单模式 + 紧急覆盖通道（安全团队可以临时授权）。
坑 3：MCP 超时
现象：Postgres MCP 偶发 30 秒超时，导致整个 Agent 任务失败。
修复：MCP 加超时重试 + 熔断器（连续 3 次超时则熔断 5 分钟）。
坑 4：Token 计费翻倍
现象：Agents API 的 token 计费比直接调 Chat Completions API 贵 2-3 倍。
原因：Agents API 内部维护了完整的 conversation 历史 + trace，每个回合都重传。
修复：用 prompt caching（重复 system prompt 自动缓存）+ 减少不必要的 handoff。
坑 5：结构化输出偶尔格式错误
现象：某些场景下 outputType 强制 JSON Schema 输出，但 Agent 输出不符合 schema。
修复：加 retry 机制 + JSON parser 容错（接受 95% 字段匹配的输出）。
坑 6：MCP 凭据泄露风险
现象：早期版本把 MCP 凭据写在环境变量里，意外泄露到日志。
修复：所有 MCP 凭据走 Artifactory Vault，运行时注入。
坑 7：多 Agent 状态不同步
现象：handoff 时上一个 Agent 的中间状态丢失。
修复：用 Redis 存 handoff 中间状态，所有 Agent 共享。
坑 8：Codex 子 Agent 误改文件
现象：Codex 子 Agent 在某些场景下"自作主张"修改了用户没要求改的文件。
修复：给 Codex 子 Agent 严格 instructions："只允许修改明确指定路径的文件"。
坑 9：MCP 启动顺序敏感
现象：某些 MCP 必须先启动才能注册（比如 Postgres MCP 启动慢）。
修复：加 MCP 健康检查 + 自动重试注册。
坑 10：Slack MCP 通知刷屏
现象：Agent 任务完成时调 Slack MCP 发通知，但有些场景下发了 10+ 条重复通知。
修复：加通知去重 + 摘要通知（多条合并成 1 条）。
坑 11：审计日志不完整
现象：早期版本只记录最终结果，丢失了中间推理过程。
修复：开启 Agents API 的 tracing 选项，全量记录 trace。
效果对比
6 周迁移后的实际效果：
| 维度 | 迁移前 | 迁移后 | 提升 |
|---|---|---|---|
| 工程师维护工时 | 76 小时/周 | 16 小时/周 | -79% |
| API 成本 | ¥28000/月 | ¥11200/月 | -60% |
| 客户工单分流时间 | 14 分钟 | 30 秒 | -96% |
| 代码评审时间 | 4 小时/PR | 45 秒/PR | -99% |
| 数据报告生成时间 | 3 小时 | 2 分钟 | -99% |
| 失败率 | 5.2% | 1.8% | -65% |
| 安全审计覆盖率 | 0% | 100% | +100% |
每月节省 480 工时，按工程师时薪 ¥150 计算 = 每月节省 ¥72000，远超项目投入。
一句话总结
OpenAI Agents API + MCP 不是替代你的 AI 脚本，而是给它们装上"编排大脑"——把零散脚本升级为可治理、可观测、可组合、可审计的 Agent 服务。
如果你正在用多个 AI 脚本处理内部工作流、或者被"AI 脚本维护混乱 / 安全审计缺失 / 无法组合"的问题困扰，OpenAI Agents API + MCP 是 2026 年 9 月最值得投入的方向。