API & LLM gateway

Client

AllyBuildClient wraps the platform API with bearer auth and project scoping. In a Python task script the workflow module exposes a configured instance as client — import it with from workflow import client. The TS module does not export a client — build one when you need raw calls:

client.get(path)                  # GET with auth headers
client.post(path, body)           # POST JSON
client.request("PUT", path, body=body)
client.request_drain("POST", path, body=body)  # long-running, 600s timeout

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

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

Tasks

TasksApi creates and observes tasks in the same project. Created tasks become children of the running task (parent_task_id) and appear on the project board unless hidden.

from workflow import create_task, wait_task

tid = create_task("summariser", params={"url": "https://example.com"})
result = wait_task(tid)          # blocks until the task settles; dict includes "result"
tasks = list_tasks()             # instances in this project
task = get_task(tid)             # status, current_phase, progress, error

create options

Approval feedback loop

When a sub-task enters pending_approval, provide feedback with feedback(). For agent tasks the message becomes the agent’s next user turn, and the call blocks until that turn finishes:

def on_feedback(task):
    feedback(task["id"], "Looks good — deploy to staging instead.")

result = wait_task(tid, on_feedback=on_feedback)

flush_tasks()   # wait for background on_complete threads before exiting

Agents and workspaces

agents = list_agents()           # roster: name, model, skills
workspaces = list_workspaces()   # workspaces in this project

Memories

Memories are the distilled lessons of previous runs. Two read modes: paginated listing, or semantic similarity search (results include a score field).

from allybuild_sdk import MemoriesApi

memories = MemoriesApi(client)

# Paginated listing for a meta-task
for m in memories.list_task_type_memories("summariser", limit=10):
    print(m["memory"])

# Semantic search: most relevant hits
hits = memories.list_task_type_memories("summariser", query="authentication", k=5)

# Most novel: least similar to the query
novel = memories.list_task_type_memories("summariser", query="authentication", k=5, order="least")

# Memories contributed by one specific run
run_notes = memories.get_task_memories(TASK_ID)

OKF knowledge base

Search the project’s OKF knowledge-base workspace for concepts, or run health checks (broken links, orphan concepts, missing citations):

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()   # knowledge-base health report

Project status

project_status is a reactive store proxy over the project’s shared state — dict-like reads (project_status["stage"]), iteration, and writes that survive across runs through the platform backend, degrading to in-memory mode if the backend is unreachable.

MCP tools

MCP servers are configured per project (integration tab, streamable-http only). mcp(server) resolves the config and returns a client with list_tools(), call(), and async twins alist_tools() / acall().

from workflow import mcp

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

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

# Concurrent calls inside 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 gateway

The gateway exposes OpenAI-compatible endpoints under /v1 on the platform root. Point any OpenAI SDK at it and authenticate with the task 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": "Summarize the failing test"}],
)

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"}]}'