call-deployed-crew

调用已部署的 Crew

介绍如何在自己的代码中调用运行于 CrewAI AMP 上的 crew 或 flow，以及如何编写 crew，确保这些调用始终正确。

已于 2026-10-01 根据 crewai 1.15.22 和 1.15.23 验证。 已于 2026-10-01 在 CrewAI AMP 和真实大模型上进行在线测试。 请先运行 crewai version；如果主版本号或次版本号不同，请使用 ask-docs skill 重新检查受版本影响的行。

---

1. 一目了然的调用约定

每个部署都有自己的基础 URL（https://<your-deployment>.crewai.com）和 bearer token，两者均显示在 AMP 控制面板的部署页面（Status 标签页）中。crewai deploy status 不会输出这两项。所有调用都要发送 Authorization: Bearer <token>。

调用	请求体	成功响应（在线）	文档
GET /inputs	-	{"inputs": ["audience", "topic"]}	GET /inputs
POST /kickoff	{"inputs": {"topic": "...", ...}}，以及可选的同级字段（第 4 节）	{"kickoff_id": "<uuid>"}	POST /kickoff
GET /status/{kickoff_id}	-	运行状态、进度、结果（第 5 节）	GET /status
POST /resume	参见第 5 节——请在你的部署上验证请求体	{"kickoff_id": "<new uuid>"}	POST /resume

使用 GET /inputs 检查服务是否就绪；GET / 可能返回 404。

预期的响应如下（也要处理文档中列出的状态码——API 介绍）：

请求	在线响应
token 错误	401 {"detail":"Invalid or missing authentication credentials"}
没有 Authorization 标头	401 {"detail":"Not authenticated"}
Crew kickoff 缺少必需键，发送 {"inputs": {}}、{} 或裸请求体	422 {"detail":"Missing inputs: audience, topic"}——这是普通字符串，不是 details.missing_inputs
请求体不是有效的 JSON	500 Internal Server Error
GET /status/<unknown id>	返回 200，且 "state": "NOT FOUND"——不是 404

export CREWAI_DEPLOYMENT_URL="https://<your-deployment>.crewai.com"
export CREWAI_DEPLOYMENT_TOKEN="<your-bearer-token>"

curl -s -H "Authorization: Bearer $CREWAI_DEPLOYMENT_TOKEN" "$CREWAI_DEPLOYMENT_URL/inputs"

curl -s -X POST "$CREWAI_DEPLOYMENT_URL/kickoff" \
  -H "Authorization: Bearer $CREWAI_DEPLOYMENT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"inputs": {"topic": "AI agents", "audience": "engineers"}}'

curl -s -H "Authorization: Bearer $CREWAI_DEPLOYMENT_TOKEN" \
  "$CREWAI_DEPLOYMENT_URL/status/<kickoff_id>"

规则：

• 将 URL 和 token 存放在环境变量或密钥存储中。切勿将 token 硬编码、提交到代码仓库、记录 Authorization 标头，或发送到浏览器——请从你的后端调用部署。
• 控制面板会显示组织级 Bearer Token 和权限较窄的 User Bearer Token。请使用能满足集成需求的最小权限 token。

---

2. 必需输入来自 {placeholder} 标记

Crew 需要的键是其任务 description / expected_output 以及 Agent 的 role / goal / backstory 中的 {name} 标记。GET /inputs 会列出已部署版本所需的键。请从 /inputs（或你的代码）中读取这些键，切勿猜测。

在本地，Crew.fetch_inputs() 是 crewai 对相同字段进行的扫描：

from crewai import Agent, Crew, Task

agent = Agent(role="{topic} researcher", goal="Explain {topic}", backstory="Writes for {audience}.")
task = Task(description="Research {topic} for {audience}.", expected_output="A summary of {topic}.", agent=agent)
print(sorted(Crew(agents=[agent], tasks=[task]).fetch_inputs()))  # ['audience', 'topic']

crewai 对收到的输入所做的处理（已在本地 1.15.22-1.15.23 上验证）：

你发送的内容	处理方式
每个占位符键	在第一次 LLM 调用前，将值替换到 Agent 和任务文本中
缺少占位符键	ValueError: Missing required template variable 'Template variable 'audience' not found in inputs dictionary' in description——在任何 LLM 调用前抛出
没有占位符的额外键	静默忽略；不会传入提示词
没有输入，或 inputs={}	本地：不会报错——跳过插值，字面文本 {topic} 会传给模型。已部署的 crew 会以 422 拒绝（第 1 节）
int / float / bool / list / dict 值	使用 str() 转换为字符串——在本地和线上部署中均如此（"audience": 3 会生成 AUDIENCE=3）。API 将每个输入值都定义为字符串——请向 crew 发送字符串（flows：见下文一节）

因此：

• 在调用 /kickoff 前，请在客户端使用 /inputs 校验请求，并发送所有键。不要指望空输入运行会明确报错——AMP 上会报错，但你自行调用 kickoff() 时不会。
• 提示词中的字面 JSON 示例（Reply like {"title": "..."}）不会被插值，但 fetch_inputs() 会将 "title": "..." 报告为输入名称。任何从花括号推导出的输入列表都会要求提供该值。请将输出结构放在 output_pydantic / response_model 中，而不是放在文本里的 JSON 示例中。
• 日期请以 ISO 8601 字符串发送，结构化值请以 JSON 字符串发送，并在 crew 内解析。

Flows 有所不同。 已部署 flow 的 GET /inputs 会列出其状态模型声明的字段——对于自行声明了 id 的测试模型，结果为 {"inputs": ["topic", "count", "id", "history"]}；若模型将 id 留给 Flow 处理，则不会列出该字段（参见 build-flow 技能）——这些字段都不是必需的：使用 {"inputs": {}} 启动 flow 会返回 200，并以默认状态运行。请仅发送你要设置的字段，并使用状态模型对应的 JSON 类型（列表应为列表——将 "[]" 作为字符串会导致运行状态变为 FAILED），且切勿发送 id（第 3 节）。

---

3. kickoff 请求体：inputs 是一个包装对象

错误	正确
{"topic": "AI agents", "audience": "engineers"}	{"inputs": {"topic": "AI agents", "audience": "engineers"}}
{"inputs": {"topic": "...", "meta": {...}}}	{"inputs": {"topic": "..."}, "meta": {...}}
{"inputs": {"id": "<previous-kickoff-id>", "topic": "..."}}（flows，已弃用）	{"inputs": {"topic": "..."}, "restoreFromStateId": "<previous-kickoff-id>"}
读取超时后重试 POST /kickoff	将其视为“可能已启动”；再次启动前先查找该运行

线上裸请求体的处理方式：crew 会返回 422 {"detail":"Missing inputs: audience, topic"}（与未发送输入相同）。flow 会返回 200，忽略顶层键并以默认状态运行——不会有任何提示告知你传入的值已被丢弃。

---

4. inputs 旁的可选字段

这些字段都位于 kickoff 请求体的顶层，绝不会放在 inputs 内：

字段	用途	文档
meta	自由格式的元数据（{"requestId": "..."}）；会在任务和 crew webhook 以及 /status 中的 last_executed_task.meta 里回传	POST /kickoff
taskWebhookUrl	每项任务完成后发送 POST 请求	同上
stepWebhookUrl	每次 Agent 思考或执行操作后调用	同上
crewWebhookUrl	执行完成时发送 POST 请求	同上
humanInputWebhookUrl	human_input=True 的任务等待审核时发送 POST 请求（第 5 节）	HITL Workflows
webhooks	事件流：{"events": [...], "url": "...", "realtime": false, "authentication": {"strategy": "bearer", "token": "<your-webhook-secret>"}}	Webhook Streaming
restoreFromStateId	对于使用 @persist 的 Flow：从先前运行中载入状态，并记录一次新的运行	inputs.id deprecation

实时测试中接收端收到的内容：

Webhook	负载	Authorization 标头
taskWebhookUrl	每项任务发送一次 POST：name、description、expected_output、agent、output、output_json、summary、kickoff_id、meta	无
crewWebhookUrl	成功时发送一次 POST：kickoff_id、meta、result、result_json、token_usage。若某次运行在 @before_kickoff 中失败，则不会发送	无
stepWebhookUrl	对于 Agent 没有工具的 crew，不会发送	-
webhooks（流式传输）	{"events": [{id, execution_id, timestamp, type, data}]}；realtime: false 会将 4 个事件合并到一次 POST 中，realtime: true 则每个事件发送一次 POST；运行失败时会收到 crew_kickoff_failed	Bearer <your-webhook-secret>

task、crew 和 human-input webhook 不包含密钥。请将其视为提示：先将 kickoff_id 与你发起的请求进行匹配，然后在采取行动前通过 GET /status 确认。

流式传输的事件可能会乱序到达（例如，前一个 task_completed 之前就收到 task_started）。请按 timestamp 或 data.emission_sequence 排序。使用 execution_id + id 去重。拒绝不包含你的 bearer 密钥的投递。

/kickoff 返回后，立即将 kickoff_id 与你自己的请求 ID 一同保存。它是你关联该次运行的唯一标识。

对于使用 @persist 的 Flow，该次运行的 state.id 就是它的 kickoff_id，因此 restoreFromStateId 接收的是先前运行的 kickoff_id。实测：从第 1 次运行恢复后，在新的状态 ID 下得到 count=2 history=['one','two']；使用未知 ID 时则无报错，并从默认值开始运行。已弃用的 {"inputs": {"id": "<kickoff_id>"}} 也能恢复状态，但会将状态写回旧 ID。

---

5. 轮询 /status/{kickoff_id}

线上部署会返回 Kickoff Crew 指南中的结构。API 参考展示了另一种结构；让客户端同时读取两种结构无需额外成本：

来源	状态字段	值	结果
线上（crew 和 flow）	state；status 是一条消息（"Task is Running"、FAILED 时的异常文本、SUCCESS 时为 null）	已见值：PENDING、STARTED、RUNNING、SUCCESS、FAILED、NOT FOUND	result 字符串（完成前为 null），result_json（纯文本时为 null）
Kickoff Crew 指南	相同	还列出 PAUSED、REVOKED	相同
API 参考	status	running、completed、error	result.output，错误文本在 error 中

规则：

1. 读取 state；仅当 state 缺失时才回退到 status。比较时统一转换为小写。
2. 终止状态：成功 = success / completed；失败 = failed / error / revoked；paused = 等待人工反馈。其他任何状态都表示继续轮询。
3. 先读取 result_json，再读取 result（字符串，或参考结构中的 result.output）。失败的运行会返回 result: null；原因在 status 中（例如 "ValueError: ..."）。
4. 采用退避策略进行轮询（从约 2 s 开始，逐渐增加到约 15 s，并加入随机抖动），并设置一个由你根据 crew 的正常运行时间选择的总截止时间。切勿使用无界的 while True 轮询。
5. 截止时间到达并不代表运行失败。请报告 kickoff_id，保留该 id，并稍后再次检查——不要重复发起运行。
6. Crews 会报告 progress.unit: "task"，并提供 completed / total。Flows 会报告 unit: "step"，其中 total 和 remaining 为 null；将 completed 显示为计数。
7. 未知 id 会以 HTTP 200 返回 state: "NOT FOUND"（文档称会返回 404；两种情况都要处理）。请确认你轮询的是发出该 id 的部署；在发起运行后，可容忍短暂的此类响应，之后应报错。等待人工输入的 crew 也会返回 NOT FOUND（见下文）。
8. 如果 /status 持续返回 5xx，运行本身可能已失败：尝试几次后就停止，并检查部署日志，而不是一直轮询到截止时间。

有大量运行同时进行时，优先使用 Webhook（第 4 节），而不是频繁轮询。

人工输入和 POST /resume——依赖前请先验证

在通过 API 使用 crew 的 human_input=True 前，请在自己的部署上验证完整流程：传入 humanInputWebhookUrl（这是运行正在等待的唯一信号；其请求正文包含 execution_id 和 UUID task_id），等待时处理 /status 返回的 NOT FOUND 和 PAUSED，检查部署接受的字段大小写形式（executionId/taskId 还是 snake_case），并确认恢复后的运行会到达 SUCCESS。

开展新工作时，可考虑使用带有 @human_feedback 的 Flow（Flows 中的人工反馈）。文档称 Webhook URL 不会传递到 /resume；请再次发送。

---

6. 首次调用

不要假定第一次请求会像后续请求一样快速响应，也不要假定第一次就会返回 200。请做好相应设计：

• 为第一次请求设置较宽裕的超时时间（按分钟而非秒计算），并在遇到 502 / 503 / 504 时采用退避策略重试，同时遵循 Retry-After。
• 将第一次请求设为 GET /inputs：它是只读请求，因此重试始终安全，而且无论如何你都需要它返回的键列表。
• 只有在能确定请求从未到达服务器（连接被拒绝 / DNS）或收到 503 时，才重试 POST /kickoff。读取超时、502 或 504 可能意味着运行已经启动：再次发起 kickoff 前，先查找已有运行。

在实测中，每次部署显示 Online 后，GET /inputs 都在约 0.1-0.25 s 内返回 200，未观察到 5xx。这只是一次样本，并非保证——仍应保留防御性设置。

---

7. 一个小型 Python 客户端

完整客户端（使用 httpx，约 200 行）位于 references/python-client.md。它涵盖第 2 至 6 节：根据 /inputs 验证输入，将输入封装在 {"inputs": ...} 中，将值转换为字符串，只重试安全的操作，在截止时间内采用退避策略轮询，并抛出携带 kickoff_id 的 RunFailed / RunPaused / RunTimeout。

import os
from deployed_crew_client import DeployedCrew, RunTimeout

crew = DeployedCrew(os.environ["CREWAI_DEPLOYMENT_URL"], os.environ["CREWAI_DEPLOYMENT_TOKEN"])
try:
    result = crew.run({"topic": "AI agents", "audience": "engineers"}, deadline_s=900)
except RunTimeout as e:
    print("still running, check later:", e.kickoff_id)
finally:
    crew.close()

对于 flow，调用 crew.run(inputs, flow=True)：它的 /inputs 会列出可选状态字段，且这些字段的值必须保留其 JSON 类型（第 2 节）。以上所有测试均于 2026-10-01 在实际部署环境中运行：成功、携带运行异常文本的 RunFailed、客户端的 MissingInputsError、服务端的 422，以及错误消息中不包含令牌的 401。

同一参考文档中还提供了带截止时间的 curl 轮询循环。

---

8. 编写 Crew，确保重复调用和并发调用都正确

你的部署会处理许多次 kickoff。不要假设每次调用都会获得全新的 Python 对象：模块全局变量、类属性、工具实例以及任何在导入时缓存的内容，都可能在一次运行结束后继续存在，并被下一次运行看到，或被同时运行的任务看到。前 3 行中的问题已在本地使用 1.15.22-1.15.23 复现（参见 references/concurrency-safe-crews.md）。

不要这样做	原因	应改为
在运行时（例如在 @before_kickoff 中）修改 task.description（或 Agent 的 role/goal/backstory）	crewai 会将首次看到的文本保留为模板，并从中重新插入变量。在同一个 Crew 对象上运行两次：第 2 次运行的提示词中带有第 1 次运行的客户信息，却没有自己的客户信息	在 {placeholders} 中放入每次运行各自的值，并将其作为输入传入；@before_kickoff 可以添加或规范化 inputs 并返回它们
在工具实例上保存每次运行的数据（self.seen、self.customer）	crew.copy()（kickoff_for_each 会使用）会复用相同的工具对象，因此第 2 次运行读到了第 1 次运行的数据	保持工具无状态；将每次运行各自的值作为工具参数传入，或从在 @before_kickoff 中设置的 contextvars.ContextVar 读取
在同一个 Crew 对象上同时运行两次 kickoff	并发的 akickoff() 调用中，第 2 次会抛出 RuntimeError: Executor is already running. Cannot invoke the same executor instance concurrently.。并发的 kickoff_async() 调用有时会抛出该错误，有时不会。	自行托管 Crew 时，每个请求都创建一个新的 Crew：ResearchCrew().crew().kickoff(inputs=...)
在内存或本地磁盘上存储结果、配额或“已处理”标记	重启后这些数据会消失，而且不会在不同工作进程之间共享	使用你自己管理的存储（数据库、对象存储），并在输入中传递 ID
在已部署的 Crew 中调用 input()	终端前没有人可以应答，因此运行无法完成	使用带有 @human_feedback 的 Flow，或使用 human_input=True + POST /resume，但前提是你已验证其可用（第 5 节）

在一次实际测试中，向一个已部署的 Crew 并发发起 8 次 kickoff，每次都在输出和 last_executed_task 中返回了各自的输入，没有发生交叉。这表明平台没有混淆这些运行；但这并不意味着上述做法是安全的——请遵循建议。

对于大型输入（大文件、大型 dataframe、大型知识库），应采用流式处理或分块处理，而不是一次性全部加载；部署前请在本地测量峰值内存。

并发运行时，共享的外部限制仍然适用：通过等待来遵守每分钟的 API 限制，并在共享存储中使用锁来申领每日配额。

9. 常见错误

症状	原因	修复方法
Crew：每次调用都出现 422 {"detail":"Missing inputs: audience, topic"}	请求体是 {"topic": ...}，没有 inputs 包装	发送 {"inputs": {...}}
Flow 忽略你发送的值，使用默认值运行	同样是未包装的请求体——Flow 会接受该请求体并返回 200	发送 {"inputs": {...}}
422 列出一个键	Agent 或任务文本中的 {placeholder} 没有匹配的输入	发送 GET /inputs 返回的所有键；检查是否有拼写错误或多余的花括号
客户端拒绝启动 Flow：缺少 count、history 等	Flow 的 /inputs 列出了所有已声明的状态字段，但没有任何必填字段	不要对 Flow 强制要求 /inputs；绝不要发送 id
启动时出现 500 Internal Server Error	请求体不是有效的 JSON	使用 JSON 库或 jq 构建请求体，而不是拼接字符串
本地出现 Missing required template variable '...' in description	同上，由 crewai 在调用任何大模型前引发	同上
输出中原样出现 {topic}	启动时没有提供输入或输入为空——跳过了插值，但没有报错	在客户端验证输入；对于包含占位符的 Crew，绝不要发送 {}
/inputs 列出类似 "title": "..." 的键	提示词中的 JSON 示例被识别为占位符	从提示词中移除 JSON 示例；使用 output_pydantic / response_model
401	令牌错误，或令牌来自其他部署	从此部署的 Status 选项卡复制令牌
轮询循环一直不结束	只等待某个精确字符串（"completed"），但 API 返回的是 SUCCESS；或者没有截止时间	转换为小写，同时接受这两种状态值，并设置截止时间
每次轮询都返回 "state": "NOT FOUND"	该 ID 对应的部署 URL 错误，或 Crew 正在等待人工输入	检查 URL；HITL 请参阅第 5 节
GET / 返回 404	线上部署没有根路由	使用 GET /inputs 进行健康检查
结果为空 / None	在状态终止前读取了 result，或忽略了 result_json	仅在成功状态后读取结果；先检查 result_json
重复运行	读取超时、502 或 504 后重试了 POST /kickoff	仅在连接失败和 503 时重试 POST /kickoff
首次调用超时，下一次调用成功	客户端超时时间过短，无法应对较慢的首次响应	为首次调用设置较长的超时时间，使用 Retry-After 进行重试，并从 GET /inputs 开始
某次运行的提示词显示了另一次运行的客户或主题	将每次运行的数据写入任务文本、工具实例或全局变量	第 8 节
Webhook 事件到达顺序错乱	HTTP 投递没有顺序保证	按 timestamp 或 data.emission_sequence 排序
/resume 返回 422，提示 "executionId ... Field required"	字段大小写与参考页面不同	请参阅第 5 节，并先测试完整的暂停 / 恢复流程

---

10. 检查清单

• 基础 URL 和令牌来自环境变量 / 密钥存储；令牌绝不写入日志或发送到浏览器
• 客户端先调用 GET /inputs，并拒绝使用缺少必需键的输入启动 Crew（Flow 除外）
• 启动请求体为 {"inputs": {...}}，其中的值为字符串；meta、Webhook 字段和 restoreFromStateId 与 inputs 并列
• Webhook 接收方会检查 bearer 密钥（流式 Webhook），或通过 /status 确认 kickoff_id（任务 / Crew Webhook 不携带密钥）
• 立即将 kickoff_id 与你的请求 ID 一起存储
• 轮询时先读取 state，再读取 status，并转换为小写；处理成功 / 失败 / 暂停 / NOT FOUND / 重复的 5xx；使用退避策略和整体截止时间
• 截止时间用尽时，报告中包含 kickoff_id，不要以新的启动请求重试
• 首个请求设置较长的超时时间，并在遵循 Retry-After 的前提下重试 502 / 503 / 504
• 运行时不会写入任务或 Agent 文本；每次运行的数据通过输入传递
• 工具和模块全局变量不保存每次运行的状态
• 大型输入使用流式传输或分块处理，并在部署前于本地测量峰值内存

---

参考资料

• Python 客户端 - 完整的 httpx 客户端、用法，以及带截止时间的 curl 轮询循环
• API 合约 - 每个端点、字段、状态和错误，以及对应的公开文档链接
• 并发安全的 crew - 在本地复现的共享状态故障，以及可避免这些故障的模式

相关技能：

• deploy-to-amp - 如何将项目部署到 AMP
• build-flow - @persist、restore_from_state_id、@human_feedback
• design-task - 使用 {placeholders} 编写任务文本
• test-crewai-project - 使用存根 LLM 离线运行 crew，以测试输入和提示词
• ask-docs - 如果这里的某个字段看起来已过时，可查询实时文档