核心能力组合：

1. 自定义工具 – 你的业务逻辑
2. OpenAPI Tools – 快速集成外部服务
3. MCP Tools – 社区工具生态系统
4. Agent Tools – 专家代理协作
5. 事件处理 + invocation_id – 长运行流程控制
   实际工作流示例：
6. 采购审批流程（替代人工审批）
   传统方式：
   员工提交申请 → 邮件通知 → 经理查看邮件 → 回复邮件
   → 财务处理 → 邮件通知员工 → 人工记录
   时间：2-3天
   Agent工作流：

定义工具

def check_budget(department: str, amount: float) -> dict:
“””检查预算”””
# 接数据库
return {“within_budget”: True, “remaining”: 50000}
def get_manager_approval(request_id: str) -> dict:
“””请求经理审批”””
# 关键：需要人工确认
confirmation = request_confirmation(
title=”采购审批”,
message=f”批准采购请求 {request_id}？”,
confirm_label=”批准”,
cancel_label=”拒绝”
)
return {“approved”: confirmation[“approved”]}
def process_payment(amount: float, vendor: str) -> dict:
“””处理付款”””
# 接OpenAPI工具（Stripe）
return stripe.charge(amount, vendor)

代理流程

root_agent = SequentialAgent(
name=”ProcurementWorkflow”,
sub_agents=[
# 步骤1：检查预算
Agent(name=”BudgetChecker”, tools=[check_budget]),

    # 步骤2：经理审批（可能暂停）
    Agent(name="ApprovalGate", tools=[get_manager_approval]),

    # 步骤3：处理付款
    Agent(name="PaymentProcessor", tools=[process_payment]),

    # 步骤4：发送通知
    Agent(name="Notification", tools=[EmailTool()])
]

)

执行

result = await runner.run_async(
message=”采购10台MacBook Pro，总计，IT部门”
)

自动检查预算 → 暂停等待经理审批 → 审批后自动付款 → 发送通知

时间：几分钟

2. 客户服务流程（替代人工客服）
   传统方式：
   客户发邮件 → 客服分类 → 查询订单 → 检查库存
   → 回复客户 → 生成发票 → 发送发票
   时间：24-48小时
   Agent工作流：

工具定义

def classify_issue(email_content: str) -> dict:
“””分类问题”””
# 使用LLM分类
return {“type”: “refund_request”, “priority”: “high”}
def get_order_details(order_id: str) -> dict:
“””获取订单详情”””
# MCP工具连接数据库
return {“status”: “delivered”, “amount”: 99.99}
def check_inventory(product_id: str) -> dict:
“””检查库存”””
# OpenAPI工具连接库存系统
return {“in_stock”: True, “quantity”: 50}
def process_refund(order_id: str) -> dict:
“””处理退款”””
# OpenAPI工具连接支付系统
return {“refund_id”: “rf123”, “status”: “processing”}

并行加速

parallel_team = ParallelAgent(
name=”ParallelInvestigation”,
sub_agents=[
Agent(name=”OrderLookup”, tools=[get_order_details]),
Agent(name=”InventoryCheck”, tools=[check_inventory])
]
)

完整流程

root_agent = SequentialAgent(
name=”CustomerService”,
sub_agents=[
Agent(name=”IssueClassifier”, tools=[classify_issue]),
parallel_team, # 同时查订单和库存
Agent(name=”DecisionMaker”, tools=[process_refund]),
Agent(name=”EmailSender”, tools=[EmailTool()])
]
)

执行

result = await runner.run_async(
message=”订单#12345，我想退款”
)

自动分类 → 并行查询订单和库存 → 决定退款 → 处理退款 → 发送邮件

时间：5-10分钟

3. 部署流程（替代DevOps人工操作）
   传统方式：
   开发者提交 → 代码审查 → 测试团队测试
   → DevOps手动部署 → 监控确认 → 回滚如果有问题
   时间：1-2周
   Agent工作流：

工具定义

def run_ci_tests(branch: str) -> dict:
“””运行CI测试”””
# MCP工具连接GitHub Actions
return github_actions.run_tests(branch)
def code_review(branch: str) -> dict:
“””代码审查”””
# MCP工具连接GitHub
return github.create_pr(branch, reviewers=[“senior_dev”])
def deploy_to_staging(branch: str) -> dict:
“””部署到staging”””
# OpenAPI工具连接Cloud Run
return cloud_run.deploy(branch, env=”staging”)
def verify_staging() -> dict:
“””验证staging环境”””
# MCP工具连接监控
return cloud_monitoring.check_health()
def get_prod_confirmation() -> dict:
“””生产部署确认”””
# 关键：需要人工确认
confirmation = request_confirmation(
title=”生产部署”,
message=f”staging验证通过，部署到生产环境？”,
confirm_label=”部署”,
cancel_label=”取消”
)
return {“approved”: confirmation[“approved”]}
def deploy_to_production(branch: str) -> dict:
“””部署到生产”””
return cloud_run.deploy(branch, env=”production”)

代理流程

root_agent = SequentialAgent(
name=”CDPipeline”,
sub_agents=[
# CI测试
Agent(name=”CI”, tools=[run_ci_tests]),

    # 代码审查（可能暂停等待人工审查）
    Agent(name="Review", tools=[code_review]),

    # 部署到staging
    Agent(name="Staging", tools=[deploy_to_staging]),

    # 验证staging
    Agent(name="Verify", tools=[verify_staging]),

    # 人工确认后部署生产
    Agent(name="ProdGate", tools=[get_prod_confirmation]),
    Agent(name="Production", tools=[deploy_to_production])
]

)

执行

result = await runner.run_async(
message=”部署feature/new-ui分支”
)

自动测试 → 创建PR → 审查通过后自动部署staging

→ 验证通过 → 暂停等待确认 → 确认后自动部署生产

时间：几小时（vs 传统1-2周）

4. 数据分析流程（替代数据分析师）
   传统方式：
   业务提出需求 → 数据分析师写SQL → 手动提取数据
   → 用Excel分析 → 生成报告 → 邮件发送
   时间：1-3天
   Agent工作流：

工具定义

def query_database(sql: str) -> dict:
“””查询数据库”””
# MCP工具连接PostgreSQL
return postgresql.query(sql)
def analyze_data(data: list) -> dict:
“””分析数据”””
# 代码执行工具
import pandas as pd
import numpy as np

df = pd.DataFrame(data)

analysis = {
    "mean": df.mean().to_dict(),
    "trend": "upward" if df.iloc[-1] > df.iloc[0] else "downward",
    "anomalies": detect_anomalies(df)
}

return analysis

def generate_report(analysis: dict) -> dict:
“””生成报告”””
# 使用LLM生成报告
return llm.generate(f”生成报告: {analysis}”)
def send_report(report: str, recipients: list) -> dict:
“””发送报告”””
# MCP工具连接Email
return email.send(recipients, report)

代理流程

root_agent = SequentialAgent(
name=”DataAnalysis”,
sub_agents=[
# 1. 理解需求并生成SQL
Agent(
name=”SQLGenerator”,
model=Gemini(model=”gemini-2.5-flash-lite”),
instruction=”理解业务需求，生成SQL查询”
),

    # 2. 执行查询
    Agent(name="DataFetcher", tools=[query_database]),

    # 3. 分析数据
    Agent(name="Analyzer", tools=[CodeExecutionTool()]),

    # 4. 生成报告
    Agent(name="ReportWriter", tools=[generate_report]),

    # 5. 发送报告
    Agent(name="Reporter", tools=[send_report])
]

)

执行

result = await runner.run_async(
message=”分析上个月的销售数据，找出增长最快的产品类别”
)

自动生成SQL → 查询数据库 → 分析数据 → 生成报告 → 发送

时间：10-30分钟（vs 传统1-3天）

能替代的工作类型：

传统工作	Agent工作流	效率提升
客服专员	自动分类、查询、回复	10-100x
数据分析师	自动查询、分析、报告	10-50x
DevOps	自动测试、部署、监控	5-20x
审批流程	自动检查、通知、记录	10-100x
内容创作	自动研究、写作、优化	5-20x