API 与 LLM 网关

Client

AllyBuildClient 封装平台 API，带 bearer 认证与项目范围。Python 任务脚本里从 workflow 导入预配置实例：from workflow import client。TS 模块不导出 client——需要裸调用时自行构建：

client.get(path)                  # 带 auth 头的 GET
client.post(path, body)           # POST JSON
client.request("PUT", path, body=body)
client.request_drain("POST", path, body=body)  # 长任务，600s 超时

import { AllyBuildClient, ALLYBUILD_API_URL, TASK_JWT, PROJECT_ID } from "@allybuild/sdk";

const client = new AllyBuildClient(ALLYBUILD_API_URL, TASK_JWT, PROJECT_ID);

任务

TasksApi 在同一项目内创建与观察任务。创建的任务成为当前运行的子任务（parent_task_id），除非隐藏，否则会出现在项目看板上。

from workflow import create_task, wait_task

tid = create_task("summariser", params={"url": "https://example.com"})
result = wait_task(tid)          # 阻塞到任务落定；dict 含 "result"
tasks = list_tasks()             # 本项目的任务实例
task = get_task(tid)             # status、current_phase、progress、error

create 选项

审批反馈闭环

当子任务进入 pending_approval 时，用 feedback() 提供反馈。对 agent 任务，消息会成为 agent 的下一轮用户输入，该调用会阻塞到那一轮结束：

def on_feedback(task):
    feedback(task["id"], "没问题——部署到 staging。")

result = wait_task(tid, on_feedback=on_feedback)

flush_tasks()   # 退出前等待后台 on_complete 线程

Agent 与工作区

agents = list_agents()           # 名册：name、model、skills
workspaces = list_workspaces()   # 项目内的工作区

记忆

记忆是过往运行沉淀的经验。两种读取方式：分页列表，或语义相似检索（结果含 score 字段）。

from allybuild_sdk import MemoriesApi

memories = MemoriesApi(client)

# 按元任务分页列出
for m in memories.list_task_type_memories("summariser", limit=10):
    print(m["memory"])

# 语义检索：最相关的命中
hits = memories.list_task_type_memories("summariser", query="authentication", k=5)

# 最新颖：与查询最不相似
novel = memories.list_task_type_memories("summariser", query="authentication", k=5, order="least")

# 某次运行贡献的记忆
run_notes = memories.get_task_memories(TASK_ID)

OKF 知识库

在项目的 OKF 知识库工作区中检索概念，或运行健康检查（断链、孤儿概念、缺失引用）：

from allybuild_sdk import OkfApi

okf = OkfApi(client)
hits = okf.search("authentication flow")
for h in hits:
    print(h["title"], "-", h.get("description", ""))

issues = okf.lint()   # 知识库健康报告

项目状态

project_status 是项目共享状态之上的响应式 store 代理——类 dict 读取（project_status["stage"]）、迭代，以及跨运行存活、经平台后端落盘的写入；后端不可达时静默降级为内存模式。

MCP 工具

MCP server 按项目配置（集成页，仅 streamable-http）。mcp(server) 解析配置并返回带 list_tools()、call() 及异步版本 alist_tools() / acall() 的 client。

from workflow import mcp, mcp_call

result = mcp("github").call("create_issue", {"title": "Fix login"})
print(result["content"][0]["text"])

tools = mcp("github").list_tools()

# async run() 内并发调用
import asyncio
r1, r2 = await asyncio.gather(
    mcp("exa").acall("web_search_exa", {"query": "MCP news"}),
    mcp("context7").acall("resolve-library-id", {"query": "fastapi", "libraryName": "fastapi"}),
)

LLM 网关

网关在平台根路径的 /v1 下暴露 OpenAI 兼容端点。把任何 OpenAI SDK 指向它，用任务 JWT 认证。

from openai import OpenAI

gateway = f"{ALLYBUILD_API_URL.removesuffix('/api')}/v1"
llm = OpenAI(base_url=gateway, api_key=TASK_JWT)

resp = llm.chat.completions.create(
    model="deepseek-chat",
    messages=[{"role": "user", "content": "总结失败的测试"}],
)

curl "${ALLYBUILD_API_URL%/api}/v1/chat/completions" \
  -H "Authorization: Bearer $TASK_JWT" \
  -H "Content-Type: application/json" \
  -d '{"model": "deepseek-chat", "messages": [{"role": "user", "content": "Hi"}]}'