Skip to main content

Plugin & Workflow Integration

Connect ColabHive to workflow tools and AI agents using the Actions API.

Quick Start​

Three steps to integrate any tool with ColabHive:

# 1. Discover — what actions are available?
GET /api/builder/v1/actions?kind=llm

# 2. Invoke — execute by slug (no UUIDs needed)
POST /api/builder/v1/actions/qwen-2.5-7b-instruct-public:invoke
Body: {"input": {"messages": [{"role": "user", "content": "Hello"}]}}

# 3. Poll — check async results
GET /api/builder/v1/invocations/{invocation_id}

n8n Integration​

HTTP Request Node​

Use n8n's HTTP Request node to call ColabHive actions directly:

Discovery (List Available Models):

  • Method: GET
  • URL: https://api.colabhive.com/api/builder/v1/actions?kind=llm
  • Headers: X-API-Key: {{$credentials.colabhiveApiKey}}

Invoke a Model:

  • Method: POST
  • URL: https://api.colabhive.com/api/builder/v1/actions/qwen-2.5-7b-instruct-public:invoke
  • Headers: X-API-Key, Content-Type: application/json
  • Body:
{
  "input": {"messages": [{"role": "user", "content": "{{$json.prompt}}"}]},
  "sync": true,
  "sync_timeout_s": 30
}

Poll Async Results (for cold models):

  • Method: GET
  • URL: https://api.colabhive.com/api/builder/v1/invocations/{{$json.invocation_id}}
  • Use n8n's Wait node + IF node to loop until status === "succeeded"

n8n Workflow Pattern​

Trigger → HTTP (discover) → Set Best Model → HTTP (invoke) → IF sync?
  ├─ Yes → Use result
  └─ No  → Wait 5s → HTTP (poll) → IF done? → Loop or Use result

AI Agent Integration (MCP / Tool-Use)​

Using Actions as AI Tools​

The Actions API is designed for AI agents that need to discover and call tools dynamically. Each action's slug is the endpoint's endpoint_name — kebab-case, DB-driven (e.g. qwen-2.5-7b-instruct-public, embeddings-public, web-fetch-public). Discover the current set with GET /api/builder/v1/actions (filter with ?kind=llm, ?task_type=…, or ?search=…) rather than hardcoding a list. This is a different naming scheme from the MCP server's built-in management tools, which are snake_case (list_endpoints, run_inference, …) — see the MCP tools reference.

import httpx

BASE = "https://api.colabhive.com/api/builder/v1"
HEADERS = {"X-API-Key": KEY}

# Agent discovers available tools
actions = httpx.get(f"{BASE}/actions", headers=HEADERS).json()

# Convert to tool definitions (e.g., for LangChain, Claude tool-use)
tools = []
for action in actions["actions"]:
    if action["input_schema"]:
        tools.append({
            "name": action["slug"],
            "description": action["description"] or action["display_name"],
            "input_schema": action["input_schema"],
        })

# Agent invokes a tool
def invoke_colabhive(slug: str, input_data: dict) -> dict:
    resp = httpx.post(
        f"{BASE}/actions/{slug}:invoke",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"input": input_data, "sync": True, "sync_timeout_s": 30},
        timeout=60,
    ).json()
    
    if resp["status"] == "succeeded":
        return resp["result"]
    
    # Poll for async
    import time
    for _ in range(30):
        status = httpx.get(
            f"{BASE}/invocations/{resp['invocation_id']}",
            headers=HEADERS,
        ).json()
        if status["status"] == "succeeded":
            return status["result"]
        if status["status"] == "failed":
            raise Exception(status["error"])
        time.sleep(2)
    raise TimeoutError("Invocation timed out")

MCP Server (Model Context Protocol)​

A standalone MCP server can expose ColabHive actions as tools for Claude, Cursor, and other MCP-compatible agents:

# Minimal MCP server concept (~50 lines)
# Each action becomes an MCP tool with its input_schema

from mcp.server import Server
import httpx

server = Server("colabhive")
BASE = "https://api.colabhive.com/api/builder/v1"

@server.list_tools()
async def list_tools():
    actions = httpx.get(f"{BASE}/actions", headers=HEADERS).json()
    return [
        {
            "name": a["slug"],
            "description": a["description"] or a["display_name"],
            "inputSchema": a["input_schema"] or {"type": "object"},
        }
        for a in actions["actions"]
        if a["input_schema"]
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    resp = httpx.post(
        f"{BASE}/actions/{name}:invoke",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"input": arguments, "sync": True},
    ).json()
    return resp.get("result", resp)

Input Schemas by Action Kind​

The Actions API auto-generates input schemas when models don't have explicit ones:

KindSchema SourceExample Input
LLMFixed: chat completion format{"messages": [{"role": "user", "content": "..."}]}
SpecialistSet at registration{"text": "..."} / {"query": "...", "documents": [...]}
ToolSet at registration{"url": "..."} / {"query": "..."}
Trained (NLP)From text_column in hyperparameters{"features": {"text": "..."}}
Trained (tabular)From target_column in hyperparameters{"features": {"col1": 1.0, "col2": "val"}}
Trained (time series)From horizon_len in hyperparameters{"values": [1.2, 3.4, ...], "horizon": 10}
GenerativeSet at registration{"prompt": "...", "width": 1024, "height": 1024}

Training Lineage​

Trained model actions include lineage information from the training job:

{
  "slug": "bert-age-classifier",
  "kind": "trained_model",
  "base_model": "bert-classification-gpu",
  "training": {
    "job_id": "uuid",
    "dataset": "demographics-383k",
    "dataset_domain": "text",
    "architecture": "bert-classification-gpu (transformers)",
    "columns": {
      "text_column": "text",
      "label_column": "age_group"
    }
  }
}

This tells a workflow tool:

  • What the model does (classifies text into age groups)
  • How it was trained (BERT on demographics data)
  • What input it expects (a text field)

Sync vs Async​

ScenarioUse sync: trueUse sync: false
Model is warm (loaded in GPU)Response in < 1sUnnecessary overhead
Model is cached (on disk)Response in 5-30sBetter for timeouts < 30s
Model is cold (not downloaded)Will timeout after 30sRequired — poll /invocations/{id}
Batch processingToo slowSend all, poll in parallel
tip

Use sync: true (default) for most cases. The API automatically returns a poll_url when the model needs loading, so you can switch to polling on the fly.

See Also​