diff --git a/CHANGELOG.md b/CHANGELOG.md
index 510d653..642fd45 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,20 @@ versioning: [Semantic Versioning](https://semver.org/).
## [Unreleased]
+### Added
+
+- **AI node** — a new flow node that calls an LLM (Claude / OpenAI / Kimi)
+ from inside a DAG. Provider, API keys, and default models resolve from the
+ project secret vault exactly like the in-app assistant
+ (`CLAUDE_API_KEY` / `OPENAI_API_KEY` / `KIMI_API_KEY`,
+ `*_DEFAULT_MODEL`, `CLAUDE_DEFAULT_PROVIDER`), or per-node overrides.
+ Prompts support full `${{...}}` variable substitution; an optional
+ JSON-object response mode parses the output so downstream nodes can read
+ `${{nodeId.json.field}}`. Output also carries `model`, `provider`,
+ `stopReason`, and token `usage`. Runs through the engine's existing retry
+ and circuit-breaker machinery. New engine package `internal/llm`
+ (standard-library only, no vendor SDK).
+
## [0.1.3] — Patch release
Two improvements that surfaced while validating v0.1.2 against a clean
diff --git a/CLAUDE.md b/CLAUDE.md
index 2868998..c165501 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -80,7 +80,7 @@ Real-time execution updates: browser connects directly to Go engine at `ws://hos
### Flow Editor
- `src/components/flow/FlowCanvas.tsx` — React Flow surface (uses `reactflow` v11).
-- `src/components/flow/nodes/` — 23 node types built on shared `BaseNode` with `iconMap`/`colorMap`. Adding a node = component in `nodes/`, register in `nodes/index.ts`, properties editor in `properties/`, and (if the engine should execute it) a case in `internal/executor/flow_executor.go`.
+- `src/components/flow/nodes/` — 25 node types built on shared `BaseNode` with `iconMap`/`colorMap`. Adding a node = component in `nodes/`, register in `nodes/index.ts`, properties editor in `properties/`, and (if the engine should execute it) a case in `internal/executor/flow_executor.go`. The AI node (LLM completion) is dispatched to `internal/llm` from `flow_executor.go`.
- `src/components/flow/properties/` — Per-node-type property editors.
- `src/components/Combobox.tsx` — In-house searchable dropdown (no third-party lib). Used everywhere instead of native `
-Visual DAG editor. Multi-runtime code execution. Pluggable queue backends.
-Encrypted secret vault. WebSocket-streamed executions. Project-scoped
-multi-tenant. All in three containers, under 140 MB RAM idle, AGPL-3.0.
+Visual DAG editor. Multi-runtime code execution. In-flow AI/LLM nodes.
+Pluggable queue backends. Encrypted secret vault. WebSocket-streamed
+executions. Project-scoped multi-tenant. All in three containers, under
+140 MB RAM idle, AGPL-3.0.
```
┌──────────────┐ drag-drop ┌──────────────┐ gocron + pg/redis/kafka ┌─────────────┐
@@ -60,6 +61,7 @@ real-time over WebSocket.
|---|---|---|---|---|---|
| Drag-drop visual DAG editor | ✅ | ❌ form-based | ❌ | ❌ Python-defined | ✅ |
| Multi-runtime code execution (Python · Node · Shell · Docker) | ✅ all 4 | ⚠️ via plugins | Shell only | Python-first | ✅ via plugins |
+| AI / LLM node in a step (Claude · OpenAI · Kimi) | ✅ built-in | ❌ | ❌ | ⚠️ provider pkg + code | ❌ |
| Pluggable queue backends (Postgres · RabbitMQ · Kafka · Redis) | ✅ 4 backends | ❌ | ❌ | ⚠️ via Celery only | ❌ |
| Built-in AES-256-GCM secret vault | ✅ | ❌ | ❌ | ⚠️ Connections (Fernet) | ❌ DB-stored |
| WebSocket per-node execution stream | ✅ | ⚠️ live log tail | ❌ | ❌ poll | ❌ poll |
@@ -71,11 +73,26 @@ real-time over WebSocket.
## Features
-- **24 built-in node types** — Start/End, Condition, Switch, Loop (for-each / batch / parallel),
+- **25 built-in node types** — Start/End, Condition, Switch, Loop (for-each / batch / parallel),
Transform, Merge, Delay, Set Variable, Sub-flow, Log, HTTP, Database (Postgres/MySQL),
Shell, Python, Node.js, Docker, Slack, Email (SMTP), Webhook (signed outbound),
Wait Webhook (inbound park), Enqueue (push to a queue), Notify (publish to a channel),
+ AI (LLM completion — Claude / OpenAI / Kimi, with JSON-mode output),
Plugin (custom user-defined).
+- **🤖 AI node — drop an LLM into any flow, no glue code.** Summarize a
+ payload, classify an incoming webhook, extract fields, or draft a reply in a
+ single node. Pick Claude, OpenAI, or Kimi (keys live in the same encrypted
+ vault as the assistant — never in the flow), template the prompt with
+ upstream data (`${{input.body}}`), and flip on **JSON mode** to get
+ structured output the next node can branch on:
+
+ ```
+ Webhook ─▶ AI node ─▶ Switch ─┬─▶ Slack (severity == "high")
+ classify to JSON └─▶ Queue (everything else)
+ ```
+
+ Token usage rides on every result, and the call retries through the same
+ circuit-breaker as every other node. No model SDK, no extra service.
- **Variable substitution everywhere** — `${{nodeId.field}}`, `${{input.X}}`,
`${{env.X}}`, `${{secrets.X}}`, plus `${{NOW}}` / `${{TODAY}}` and
`${{loop.item}}` inside loop bodies.
@@ -221,6 +238,9 @@ For deeper dives see [`CLAUDE.md`](./CLAUDE.md) and [`docs/development/SETUP.md`
## Use cases we've seen
- **Cron → external API → Slack alert** — three nodes, two minutes.
+- **AI triage** — Inbound webhook → **AI node** classifies the payload to JSON
+ (`{ "severity": "high", "team": "billing" }`) → Switch routes high-severity
+ to a Slack alert and the rest to a queue. Five nodes, no model SDK.
- **Postgres ETL** — DB query → Transform → DB upsert, on a schedule, with
a retry policy and a dead-letter queue.
- **Webhook fan-out** — Inbound webhook (`WAIT_WEBHOOK`) parks until your
diff --git a/apps/runloop-engine/internal/executor/flow_executor.go b/apps/runloop-engine/internal/executor/flow_executor.go
index c579606..17e7670 100644
--- a/apps/runloop-engine/internal/executor/flow_executor.go
+++ b/apps/runloop-engine/internal/executor/flow_executor.go
@@ -23,6 +23,7 @@ import (
"github.com/runloop/runloop-engine/internal/connector"
"github.com/runloop/runloop-engine/internal/db"
"github.com/runloop/runloop-engine/internal/idgen"
+ "github.com/runloop/runloop-engine/internal/llm"
"github.com/runloop/runloop-engine/internal/logging"
"github.com/runloop/runloop-engine/internal/models"
"github.com/runloop/runloop-engine/internal/worker"
@@ -659,6 +660,9 @@ func (fe *FlowExecutor) dispatchNode(
case "SLACK", "EMAIL", "WEBHOOK":
return fe.executeConnectorNode(ctx, string(node.Type), config)
+
+ case models.JobTypeAI:
+ return fe.executeAINode(ctx, task, config)
}
// Fall through to plugin registry. The node's type string becomes the
@@ -1438,6 +1442,186 @@ func (fe *FlowExecutor) executeConnectorNode(ctx context.Context, connType strin
}, nil
}
+// aiModelSecret maps a provider to the secret name that holds its default
+// model override, mirroring the Next.js assistant proxy.
+var aiModelSecret = map[llm.Provider]string{
+ llm.ProviderClaude: "CLAUDE_DEFAULT_MODEL",
+ llm.ProviderOpenAI: "OPENAI_DEFAULT_MODEL",
+ llm.ProviderKimi: "KIMI_DEFAULT_MODEL",
+}
+
+var aiKeySecret = map[llm.Provider]string{
+ llm.ProviderClaude: "CLAUDE_API_KEY",
+ llm.ProviderOpenAI: "OPENAI_API_KEY",
+ llm.ProviderKimi: "KIMI_API_KEY",
+}
+
+// executeAINode runs an AI/LLM completion node. Config (post substitution):
+//
+// prompt string required — the user message; ${{...}} already resolved
+// systemPrompt string optional — system steer
+// provider string optional — claude | openai | kimi (else auto-detect)
+// model string optional — overrides the provider default
+// apiKey string optional — explicit key (e.g. ${{secrets.X}}); else vault
+// maxTokens number optional — capped at llm.MaxTokensCap
+// temperature number optional
+// responseFormat string optional — "json" asks for a single JSON object
+// timeout number optional — seconds; default 60
+//
+// Output: response, model, provider, stopReason, usage{...}, and (json mode)
+// a parsed `json` object. Reference downstream as ${{nodeId.response}},
+// ${{nodeId.json.field}}, ${{nodeId.usage.totalTokens}}.
+func (fe *FlowExecutor) executeAINode(ctx context.Context, task *worker.Task, config models.JSONMap) (*models.JobResult, error) {
+ prompt, _ := config["prompt"].(string)
+ if strings.TrimSpace(prompt) == "" {
+ if m, ok := config["message"].(string); ok {
+ prompt = m
+ }
+ }
+ if strings.TrimSpace(prompt) == "" {
+ return &models.JobResult{
+ Success: false,
+ ErrorMessage: strPtr("AI node: 'prompt' is required"),
+ }, nil
+ }
+
+ getSecret := func(name string) string {
+ if fe.secretStore == nil {
+ return ""
+ }
+ v, err := fe.secretStore.GetSecret(ctx, name, task.ProjectID)
+ if err != nil {
+ return ""
+ }
+ return v
+ }
+
+ // --- provider + key resolution (mirrors the assistant proxy) ---
+ keys := map[llm.Provider]string{
+ llm.ProviderClaude: getSecret(aiKeySecret[llm.ProviderClaude]),
+ llm.ProviderOpenAI: getSecret(aiKeySecret[llm.ProviderOpenAI]),
+ llm.ProviderKimi: getSecret(aiKeySecret[llm.ProviderKimi]),
+ }
+
+ var provider llm.Provider
+ if p, _ := config["provider"].(string); strings.TrimSpace(p) != "" {
+ provider = llm.Provider(strings.ToLower(strings.TrimSpace(p)))
+ } else if pref := strings.ToLower(strings.TrimSpace(getSecret("CLAUDE_DEFAULT_PROVIDER"))); pref != "" {
+ provider = llm.Provider(pref)
+ }
+ switch provider {
+ case llm.ProviderClaude, llm.ProviderOpenAI, llm.ProviderKimi:
+ // explicit/preferred provider is valid
+ default:
+ // auto-detect: first provider with a configured key
+ switch {
+ case keys[llm.ProviderClaude] != "":
+ provider = llm.ProviderClaude
+ case keys[llm.ProviderOpenAI] != "":
+ provider = llm.ProviderOpenAI
+ case keys[llm.ProviderKimi] != "":
+ provider = llm.ProviderKimi
+ default:
+ return &models.JobResult{
+ Success: false,
+ ErrorMessage: strPtr("AI node: no provider configured — set CLAUDE_API_KEY, OPENAI_API_KEY, or KIMI_API_KEY (or a node 'provider')"),
+ }, nil
+ }
+ }
+
+ apiKey, _ := config["apiKey"].(string)
+ if strings.TrimSpace(apiKey) == "" {
+ apiKey = keys[provider]
+ }
+ if strings.TrimSpace(apiKey) == "" {
+ return &models.JobResult{
+ Success: false,
+ ErrorMessage: strPtr(fmt.Sprintf("AI node: provider %q selected but %s is not set in this project's secrets", provider, aiKeySecret[provider])),
+ }, nil
+ }
+
+ // --- model + tuning ---
+ model, _ := config["model"].(string)
+ if strings.TrimSpace(model) == "" {
+ model = getSecret(aiModelSecret[provider])
+ }
+
+ maxTokens := 1024
+ if v, ok := config["maxTokens"].(float64); ok && v > 0 {
+ maxTokens = int(v)
+ }
+
+ var temperature *float64
+ if v, ok := config["temperature"].(float64); ok {
+ t := v
+ temperature = &t
+ }
+
+ jsonMode := false
+ if rf, _ := config["responseFormat"].(string); strings.EqualFold(rf, "json") {
+ jsonMode = true
+ }
+
+ timeout := 60 * time.Second
+ if v, ok := config["timeout"].(float64); ok && v > 0 {
+ timeout = time.Duration(v) * time.Second
+ }
+
+ system, _ := config["systemPrompt"].(string)
+ if system == "" {
+ system, _ = config["system"].(string)
+ }
+
+ client := llm.New(&http.Client{Timeout: timeout})
+ resp, err := client.Complete(ctx, llm.Request{
+ Provider: provider,
+ APIKey: apiKey,
+ Model: model,
+ System: system,
+ Messages: []llm.Message{{Role: "user", Content: prompt}},
+ MaxTokens: maxTokens,
+ Temperature: temperature,
+ JSONMode: jsonMode,
+ })
+ if err != nil {
+ return &models.JobResult{
+ Success: false,
+ ErrorMessage: strPtr(err.Error()),
+ }, nil
+ }
+
+ output := models.JSONMap{
+ "response": resp.Text,
+ "model": resp.Model,
+ "provider": string(resp.Provider),
+ "stopReason": resp.StopReason,
+ "usage": map[string]interface{}{
+ "promptTokens": resp.Usage.PromptTokens,
+ "completionTokens": resp.Usage.CompletionTokens,
+ "totalTokens": resp.Usage.TotalTokens,
+ },
+ }
+
+ // In JSON mode, surface the parsed object as `json` so downstream nodes can
+ // reference ${{nodeId.json.field}}. A parse failure isn't fatal — the raw
+ // text is still available on `response` — but we flag it in the logs.
+ parseNote := ""
+ if jsonMode {
+ var parsed interface{}
+ if err := json.Unmarshal([]byte(strings.TrimSpace(resp.Text)), &parsed); err == nil {
+ output["json"] = parsed
+ } else {
+ parseNote = " (responseFormat=json but output was not valid JSON)"
+ }
+ }
+
+ return &models.JobResult{
+ Success: true,
+ Output: output,
+ Logs: fmt.Sprintf("%s/%s · %d prompt + %d completion = %d tokens · stop=%s%s", resp.Provider, resp.Model, resp.Usage.PromptTokens, resp.Usage.CompletionTokens, resp.Usage.TotalTokens, resp.StopReason, parseNote),
+ }, nil
+}
+
func (fe *FlowExecutor) resolveSecrets(ctx context.Context, config models.JSONMap, projectID string) (models.JSONMap, error) {
result := make(models.JSONMap)
for key, value := range config {
diff --git a/apps/runloop-engine/internal/llm/client.go b/apps/runloop-engine/internal/llm/client.go
new file mode 100644
index 0000000..1b2cc23
--- /dev/null
+++ b/apps/runloop-engine/internal/llm/client.go
@@ -0,0 +1,318 @@
+// Package llm is a tiny, dependency-free client for the three chat providers
+// RunLoop supports: Claude (Anthropic Messages API), OpenAI, and Kimi
+// (Moonshot, OpenAI-compatible). It is the engine-side twin of the Next.js
+// AI assistant proxy (apps/runloop/src/app/api/ai/chat/route.ts) — same
+// provider set, same secret names, same default models — so a flow's AI node
+// behaves identically to the in-app assistant.
+//
+// The package deliberately depends only on the standard library: the AI node
+// is on the flow execution hot path and we don't want a vendor SDK pulling in
+// its own retry/transport behavior underneath the engine's own retry +
+// circuit-breaker machinery.
+package llm
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "strings"
+)
+
+// Provider identifies a chat backend. Values match the strings stored in the
+// CLAUDE_DEFAULT_PROVIDER secret and accepted by the AI node's `provider`
+// field.
+type Provider string
+
+const (
+ ProviderClaude Provider = "claude"
+ ProviderOpenAI Provider = "openai"
+ ProviderKimi Provider = "kimi"
+)
+
+// DefaultModel is the per-provider fallback model, used when neither the node
+// config nor a _DEFAULT_MODEL secret specifies one. Kept in sync
+// with DEFAULT_MODEL in the Next.js route.
+var DefaultModel = map[Provider]string{
+ ProviderClaude: "claude-sonnet-4-7",
+ ProviderOpenAI: "gpt-4o-mini",
+ ProviderKimi: "kimi-latest",
+}
+
+// MaxTokensCap is the hard ceiling on completion length, matching the
+// assistant proxy. Defends against a runaway flow burning a provider budget.
+const MaxTokensCap = 4096
+
+// Provider endpoints. Vars (not consts) so tests can point them at an
+// httptest server; never reassigned in production code.
+var (
+ anthropicEndpoint = "https://api.anthropic.com/v1/messages"
+ openaiEndpoint = "https://api.openai.com/v1/chat/completions"
+ kimiEndpoint = "https://api.moonshot.cn/v1/chat/completions"
+)
+
+// Message is one turn of the conversation. The AI node only ever sends a
+// single user message today, but the slice keeps the door open for an
+// agent/tool loop later.
+type Message struct {
+ Role string `json:"role"` // "user" | "assistant"
+ Content string `json:"content"`
+}
+
+// Request is a provider-agnostic completion request.
+type Request struct {
+ Provider Provider
+ APIKey string
+ Model string
+ System string
+ Messages []Message
+ MaxTokens int
+ Temperature *float64 // pointer so "unset" is distinguishable from 0
+ JSONMode bool // ask the model to emit a single JSON object
+}
+
+// Usage reports token counts, normalized across providers.
+type Usage struct {
+ PromptTokens int `json:"promptTokens"`
+ CompletionTokens int `json:"completionTokens"`
+ TotalTokens int `json:"totalTokens"`
+}
+
+// Response is the normalized result.
+type Response struct {
+ Text string `json:"text"`
+ Model string `json:"model"`
+ Provider Provider `json:"provider"`
+ StopReason string `json:"stopReason"`
+ Usage Usage `json:"usage"`
+}
+
+// Client performs completions. The zero value is unusable; construct with New.
+type Client struct {
+ http *http.Client
+}
+
+// New builds a Client over the given http.Client. Pass an *http.Client with
+// the timeout you want the LLM call bounded by — the engine sizes this from
+// the node's timeout config.
+func New(httpClient *http.Client) *Client {
+ if httpClient == nil {
+ httpClient = http.DefaultClient
+ }
+ return &Client{http: httpClient}
+}
+
+// Complete dispatches to the right provider and returns a normalized response.
+func (c *Client) Complete(ctx context.Context, req Request) (*Response, error) {
+ if req.APIKey == "" {
+ return nil, fmt.Errorf("llm: API key is empty for provider %q", req.Provider)
+ }
+ if len(req.Messages) == 0 {
+ return nil, fmt.Errorf("llm: at least one message is required")
+ }
+ if req.MaxTokens <= 0 {
+ req.MaxTokens = 1024
+ }
+ if req.MaxTokens > MaxTokensCap {
+ req.MaxTokens = MaxTokensCap
+ }
+ if req.Model == "" {
+ req.Model = DefaultModel[req.Provider]
+ }
+
+ switch req.Provider {
+ case ProviderClaude:
+ return c.completeClaude(ctx, req)
+ case ProviderOpenAI:
+ return c.completeOpenAICompatible(ctx, req, openaiEndpoint)
+ case ProviderKimi:
+ return c.completeOpenAICompatible(ctx, req, kimiEndpoint)
+ default:
+ return nil, fmt.Errorf("llm: unknown provider %q", req.Provider)
+ }
+}
+
+// ---- Anthropic (Claude) ----
+
+func (c *Client) completeClaude(ctx context.Context, req Request) (*Response, error) {
+ system := req.System
+ if req.JSONMode {
+ // Claude has no native JSON-mode param, so steer via the system text.
+ hint := "Respond with a single valid JSON object and nothing else — no prose, no markdown fences."
+ if system == "" {
+ system = hint
+ } else {
+ system = system + "\n\n" + hint
+ }
+ }
+
+ payload := map[string]interface{}{
+ "model": req.Model,
+ "max_tokens": req.MaxTokens,
+ "messages": req.Messages,
+ }
+ if system != "" {
+ payload["system"] = system
+ }
+ if req.Temperature != nil {
+ payload["temperature"] = *req.Temperature
+ }
+
+ httpReq, err := c.newJSONRequest(ctx, anthropicEndpoint, payload)
+ if err != nil {
+ return nil, err
+ }
+ httpReq.Header.Set("x-api-key", req.APIKey)
+ httpReq.Header.Set("anthropic-version", "2023-06-01")
+
+ body, err := c.do(httpReq, ProviderClaude)
+ if err != nil {
+ return nil, err
+ }
+
+ var parsed struct {
+ Model string `json:"model"`
+ StopReason string `json:"stop_reason"`
+ Content []struct {
+ Type string `json:"type"`
+ Text string `json:"text"`
+ } `json:"content"`
+ Usage struct {
+ InputTokens int `json:"input_tokens"`
+ OutputTokens int `json:"output_tokens"`
+ } `json:"usage"`
+ }
+ if err := json.Unmarshal(body, &parsed); err != nil {
+ return nil, fmt.Errorf("llm: failed to parse Claude response: %w", err)
+ }
+
+ var text strings.Builder
+ for _, block := range parsed.Content {
+ if block.Type == "text" {
+ text.WriteString(block.Text)
+ }
+ }
+
+ return &Response{
+ Text: text.String(),
+ Model: orDefault(parsed.Model, req.Model),
+ Provider: ProviderClaude,
+ StopReason: parsed.StopReason,
+ Usage: Usage{
+ PromptTokens: parsed.Usage.InputTokens,
+ CompletionTokens: parsed.Usage.OutputTokens,
+ TotalTokens: parsed.Usage.InputTokens + parsed.Usage.OutputTokens,
+ },
+ }, nil
+}
+
+// ---- OpenAI / Kimi (chat completions) ----
+
+func (c *Client) completeOpenAICompatible(ctx context.Context, req Request, endpoint string) (*Response, error) {
+ messages := make([]map[string]string, 0, len(req.Messages)+1)
+ if req.System != "" {
+ messages = append(messages, map[string]string{"role": "system", "content": req.System})
+ }
+ for _, m := range req.Messages {
+ messages = append(messages, map[string]string{"role": m.Role, "content": m.Content})
+ }
+
+ payload := map[string]interface{}{
+ "model": req.Model,
+ "messages": messages,
+ "max_tokens": req.MaxTokens,
+ }
+ if req.Temperature != nil {
+ payload["temperature"] = *req.Temperature
+ }
+ if req.JSONMode {
+ payload["response_format"] = map[string]string{"type": "json_object"}
+ }
+
+ httpReq, err := c.newJSONRequest(ctx, endpoint, payload)
+ if err != nil {
+ return nil, err
+ }
+ httpReq.Header.Set("Authorization", "Bearer "+req.APIKey)
+
+ body, err := c.do(httpReq, req.Provider)
+ if err != nil {
+ return nil, err
+ }
+
+ var parsed struct {
+ Model string `json:"model"`
+ Choices []struct {
+ Message struct{ Content string `json:"content"` } `json:"message"`
+ FinishReason string `json:"finish_reason"`
+ } `json:"choices"`
+ Usage struct {
+ PromptTokens int `json:"prompt_tokens"`
+ CompletionTokens int `json:"completion_tokens"`
+ TotalTokens int `json:"total_tokens"`
+ } `json:"usage"`
+ }
+ if err := json.Unmarshal(body, &parsed); err != nil {
+ return nil, fmt.Errorf("llm: failed to parse %s response: %w", req.Provider, err)
+ }
+ if len(parsed.Choices) == 0 {
+ return nil, fmt.Errorf("llm: %s returned no choices", req.Provider)
+ }
+
+ return &Response{
+ Text: parsed.Choices[0].Message.Content,
+ Model: orDefault(parsed.Model, req.Model),
+ Provider: req.Provider,
+ StopReason: parsed.Choices[0].FinishReason,
+ Usage: Usage{
+ PromptTokens: parsed.Usage.PromptTokens,
+ CompletionTokens: parsed.Usage.CompletionTokens,
+ TotalTokens: parsed.Usage.TotalTokens,
+ },
+ }, nil
+}
+
+// ---- shared transport ----
+
+func (c *Client) newJSONRequest(ctx context.Context, endpoint string, payload interface{}) (*http.Request, error) {
+ buf, err := json.Marshal(payload)
+ if err != nil {
+ return nil, fmt.Errorf("llm: failed to encode request: %w", err)
+ }
+ httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(buf))
+ if err != nil {
+ return nil, fmt.Errorf("llm: failed to build request: %w", err)
+ }
+ httpReq.Header.Set("Content-Type", "application/json")
+ return httpReq, nil
+}
+
+func (c *Client) do(req *http.Request, provider Provider) ([]byte, error) {
+ resp, err := c.http.Do(req)
+ if err != nil {
+ return nil, fmt.Errorf("llm: %s request failed: %w", provider, err)
+ }
+ defer resp.Body.Close()
+
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ return nil, fmt.Errorf("llm: failed to read %s response: %w", provider, err)
+ }
+ if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+ snippet := string(body)
+ if len(snippet) > 500 {
+ snippet = snippet[:500]
+ }
+ return nil, fmt.Errorf("llm: %s API %d: %s", provider, resp.StatusCode, snippet)
+ }
+ return body, nil
+}
+
+func orDefault(v, fallback string) string {
+ if v == "" {
+ return fallback
+ }
+ return v
+}
diff --git a/apps/runloop-engine/internal/llm/client_test.go b/apps/runloop-engine/internal/llm/client_test.go
new file mode 100644
index 0000000..ffbbfce
--- /dev/null
+++ b/apps/runloop-engine/internal/llm/client_test.go
@@ -0,0 +1,189 @@
+package llm
+
+import (
+ "context"
+ "encoding/json"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+)
+
+func TestCompleteClaude(t *testing.T) {
+ var gotPath, gotKey, gotVersion string
+ var gotBody map[string]interface{}
+
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ gotPath = r.URL.Path
+ gotKey = r.Header.Get("x-api-key")
+ gotVersion = r.Header.Get("anthropic-version")
+ raw, _ := io.ReadAll(r.Body)
+ _ = json.Unmarshal(raw, &gotBody)
+
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{
+ "model": "claude-sonnet-4-7",
+ "stop_reason": "end_turn",
+ "content": [{"type":"text","text":"hello world"}],
+ "usage": {"input_tokens": 12, "output_tokens": 3}
+ }`))
+ }))
+ defer srv.Close()
+
+ orig := anthropicEndpoint
+ anthropicEndpoint = srv.URL
+ defer func() { anthropicEndpoint = orig }()
+
+ temp := 0.5
+ resp, err := New(srv.Client()).Complete(context.Background(), Request{
+ Provider: ProviderClaude,
+ APIKey: "sk-test",
+ System: "be terse",
+ Messages: []Message{{Role: "user", Content: "hi"}},
+ MaxTokens: 100,
+ Temperature: &temp,
+ })
+ if err != nil {
+ t.Fatalf("Complete returned error: %v", err)
+ }
+
+ if gotPath == "" || gotKey != "sk-test" || gotVersion != "2023-06-01" {
+ t.Errorf("request headers wrong: key=%q version=%q", gotKey, gotVersion)
+ }
+ if gotBody["system"] != "be terse" {
+ t.Errorf("system not forwarded: %v", gotBody["system"])
+ }
+ if gotBody["max_tokens"].(float64) != 100 {
+ t.Errorf("max_tokens wrong: %v", gotBody["max_tokens"])
+ }
+ if gotBody["temperature"].(float64) != 0.5 {
+ t.Errorf("temperature wrong: %v", gotBody["temperature"])
+ }
+ if resp.Text != "hello world" {
+ t.Errorf("text = %q, want %q", resp.Text, "hello world")
+ }
+ if resp.StopReason != "end_turn" {
+ t.Errorf("stopReason = %q", resp.StopReason)
+ }
+ if resp.Usage.TotalTokens != 15 {
+ t.Errorf("totalTokens = %d, want 15", resp.Usage.TotalTokens)
+ }
+}
+
+func TestCompleteOpenAIJSONMode(t *testing.T) {
+ var gotBody map[string]interface{}
+ var gotAuth string
+
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ gotAuth = r.Header.Get("Authorization")
+ raw, _ := io.ReadAll(r.Body)
+ _ = json.Unmarshal(raw, &gotBody)
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{
+ "model": "gpt-4o-mini",
+ "choices": [{"message": {"content": "{\"ok\":true}"}, "finish_reason": "stop"}],
+ "usage": {"prompt_tokens": 5, "completion_tokens": 4, "total_tokens": 9}
+ }`))
+ }))
+ defer srv.Close()
+
+ orig := openaiEndpoint
+ openaiEndpoint = srv.URL
+ defer func() { openaiEndpoint = orig }()
+
+ resp, err := New(srv.Client()).Complete(context.Background(), Request{
+ Provider: ProviderOpenAI,
+ APIKey: "sk-oai",
+ System: "sys",
+ Messages: []Message{{Role: "user", Content: "give me json"}},
+ MaxTokens: 50,
+ JSONMode: true,
+ })
+ if err != nil {
+ t.Fatalf("Complete returned error: %v", err)
+ }
+
+ if gotAuth != "Bearer sk-oai" {
+ t.Errorf("auth header = %q", gotAuth)
+ }
+ rf, ok := gotBody["response_format"].(map[string]interface{})
+ if !ok || rf["type"] != "json_object" {
+ t.Errorf("response_format not set for JSON mode: %v", gotBody["response_format"])
+ }
+ // system + user message should both be present, system first
+ msgs, _ := gotBody["messages"].([]interface{})
+ if len(msgs) != 2 {
+ t.Fatalf("expected 2 messages, got %d", len(msgs))
+ }
+ if first := msgs[0].(map[string]interface{}); first["role"] != "system" {
+ t.Errorf("first message role = %v, want system", first["role"])
+ }
+ if resp.Text != `{"ok":true}` {
+ t.Errorf("text = %q", resp.Text)
+ }
+ if resp.Usage.TotalTokens != 9 {
+ t.Errorf("totalTokens = %d, want 9", resp.Usage.TotalTokens)
+ }
+}
+
+func TestCompleteMaxTokensCapped(t *testing.T) {
+ var gotBody map[string]interface{}
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ raw, _ := io.ReadAll(r.Body)
+ _ = json.Unmarshal(raw, &gotBody)
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{"content":[{"type":"text","text":"x"}],"usage":{}}`))
+ }))
+ defer srv.Close()
+
+ orig := anthropicEndpoint
+ anthropicEndpoint = srv.URL
+ defer func() { anthropicEndpoint = orig }()
+
+ _, err := New(srv.Client()).Complete(context.Background(), Request{
+ Provider: ProviderClaude,
+ APIKey: "k",
+ Messages: []Message{{Role: "user", Content: "hi"}},
+ MaxTokens: 999999,
+ })
+ if err != nil {
+ t.Fatalf("Complete returned error: %v", err)
+ }
+ if gotBody["max_tokens"].(float64) != float64(MaxTokensCap) {
+ t.Errorf("max_tokens = %v, want capped at %d", gotBody["max_tokens"], MaxTokensCap)
+ }
+}
+
+func TestCompleteUpstreamError(t *testing.T) {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.WriteHeader(http.StatusUnauthorized)
+ _, _ = w.Write([]byte(`{"error":"invalid key"}`))
+ }))
+ defer srv.Close()
+
+ orig := openaiEndpoint
+ openaiEndpoint = srv.URL
+ defer func() { openaiEndpoint = orig }()
+
+ _, err := New(srv.Client()).Complete(context.Background(), Request{
+ Provider: ProviderOpenAI,
+ APIKey: "bad",
+ Messages: []Message{{Role: "user", Content: "hi"}},
+ })
+ if err == nil {
+ t.Fatal("expected error on 401, got nil")
+ }
+}
+
+func TestCompleteValidation(t *testing.T) {
+ c := New(nil)
+ if _, err := c.Complete(context.Background(), Request{Provider: ProviderClaude, Messages: []Message{{Role: "user", Content: "x"}}}); err == nil {
+ t.Error("expected error for empty API key")
+ }
+ if _, err := c.Complete(context.Background(), Request{Provider: ProviderClaude, APIKey: "k"}); err == nil {
+ t.Error("expected error for no messages")
+ }
+ if _, err := c.Complete(context.Background(), Request{Provider: "bogus", APIKey: "k", Messages: []Message{{Role: "user", Content: "x"}}}); err == nil {
+ t.Error("expected error for unknown provider")
+ }
+}
diff --git a/apps/runloop-engine/internal/models/scheduler.go b/apps/runloop-engine/internal/models/scheduler.go
index 512da33..ba36e38 100644
--- a/apps/runloop-engine/internal/models/scheduler.go
+++ b/apps/runloop-engine/internal/models/scheduler.go
@@ -51,6 +51,7 @@ const (
JobTypeDocker JobType = "DOCKER"
JobTypeSlack JobType = "SLACK"
JobTypeEmail JobType = "EMAIL"
+ JobTypeAI JobType = "AI"
// Flow-shape nodes: handled directly by FlowExecutor, not by per-type executors.
JobTypeStart JobType = "START"
JobTypeEnd JobType = "END"
diff --git a/apps/runloop/src/app/(protected)/p/[projectId]/docs/page.tsx b/apps/runloop/src/app/(protected)/p/[projectId]/docs/page.tsx
index 3f6b013..b18958e 100644
--- a/apps/runloop/src/app/(protected)/p/[projectId]/docs/page.tsx
+++ b/apps/runloop/src/app/(protected)/p/[projectId]/docs/page.tsx
@@ -60,7 +60,7 @@ const RAW_ENDPOINTS: EndpointDef[] = [
{ group: 'Flows', method: 'POST', path: '/api/flows',
summary: 'Create a flow. flowConfig has nodes[] + edges[].',
body: '{\n "projectId": "demo-project",\n "name": "send-welcome",\n "type": "DAG",\n "flowConfig": { "edges": [...], "nodes": [...] }\n}',
- notes: 'Node types: START · END · CONDITION · DELAY · LOOP · TRANSFORM · MERGE · SWITCH · LOG · SET_VARIABLE · SUBFLOW · WEBHOOK_OUT · WAIT_WEBHOOK · ENQUEUE · NOTIFY · HTTP · DATABASE · SHELL · PYTHON · NODEJS · DOCKER · SLACK · EMAIL.' },
+ notes: 'Node types: START · END · CONDITION · DELAY · LOOP · TRANSFORM · MERGE · SWITCH · LOG · SET_VARIABLE · SUBFLOW · WEBHOOK_OUT · WAIT_WEBHOOK · ENQUEUE · NOTIFY · HTTP · DATABASE · SHELL · PYTHON · NODEJS · DOCKER · SLACK · EMAIL · AI.' },
// Channels
{ group: 'Channels', method: 'GET', path: '/api/channels', summary: 'Active pub/sub channels with live subscriber counts.' },
diff --git a/apps/runloop/src/components/flow/FlowCanvas.tsx b/apps/runloop/src/components/flow/FlowCanvas.tsx
index 2d8e339..b2a6a4f 100644
--- a/apps/runloop/src/components/flow/FlowCanvas.tsx
+++ b/apps/runloop/src/components/flow/FlowCanvas.tsx
@@ -24,7 +24,7 @@ import 'reactflow/dist/style.css';
import {
Play, Square, Globe, Database, Terminal, Code2, Hash,
Container as Docker, Slack, Mail, GitBranch, Clock, RotateCcw, Variable,
- GitMerge, Split, FileText, PenLine, Workflow, Webhook, Hourglass, Inbox, BellRing,
+ GitMerge, Split, FileText, PenLine, Workflow, Webhook, Hourglass, Inbox, BellRing, Sparkles,
Save, TestTube2, Wand2, Plus, Settings, X, AlertCircle, Search, ChevronDown,
} from 'lucide-react';
import { nodeTypes as importedNodeTypes } from './nodes';
@@ -51,6 +51,9 @@ const PALETTE = [
{ type: 'nodejsNode', label: 'Node.js', nodeType: 'nodejs', icon: Hash, color: '#22C55E' },
{ type: 'dockerNode', label: 'Docker', nodeType: 'docker', icon: Docker, color: '#0EA5E9' },
]},
+ { category: 'AI', items: [
+ { type: 'aiNode', label: 'AI', nodeType: 'ai', icon: Sparkles, color: '#8B5CF6' },
+ ]},
{ category: 'Notifications', items: [
{ type: 'slackNode', label: 'Slack', nodeType: 'slack', icon: Slack, color: '#A855F7' },
{ type: 'emailNode', label: 'Email', nodeType: 'email', icon: Mail, color: '#F97316' },
diff --git a/apps/runloop/src/components/flow/nodes/AINode.tsx b/apps/runloop/src/components/flow/nodes/AINode.tsx
new file mode 100644
index 0000000..24708e5
--- /dev/null
+++ b/apps/runloop/src/components/flow/nodes/AINode.tsx
@@ -0,0 +1,26 @@
+'use client';
+
+import { NodeProps } from 'reactflow';
+import { BaseNode, BaseNodeData } from './BaseNode';
+
+const PROVIDER_LABEL: Record = {
+ claude: 'Claude',
+ openai: 'OpenAI',
+ kimi: 'Kimi',
+};
+
+export function AINode(props: NodeProps) {
+ const { provider, model } = props.data.config || {};
+ const label = provider ? PROVIDER_LABEL[provider] || provider : 'auto';
+
+ return (
+
+