Hands-on code for the blog posts: a tiny AI coder + a Plan→Research→Synthesize agentic workflow, both powered by OllaBridge Cloud (the router LLM). See https://ruslanmv.com/blog/

app1_codegen.py

#!/usr/bin/env python3
"""Program 1 — "Describe it, get a running project."

A tiny coding assistant in the spirit of GitPilot: you describe a program in one
sentence, an LLM (routed through OllaBridge Cloud) writes the files, we save them
to ./generated/, and then we RUN the result so you can see it actually works.

    python app1_codegen.py "a CLI that prints the first 10 Fibonacci numbers"

It demonstrates the core loop every AI coder uses:
    describe  ->  plan + files (JSON)  ->  write to disk  ->  run + show output
"""
from __future__ import annotations

import json
import re
import subprocess
import sys
from pathlib import Path

from ollabridge_client import chat

OUT = Path(__file__).parent / "generated"

SYSTEM = (
    "You are a senior Python engineer. Given a task, return ONLY a strict JSON "
    "object (no markdown fences) with this exact shape:\n"
    '{"files": [{"path": "main.py", "content": "..."}], "run": "python main.py"}\n'
    "Keep it to a single self-contained file named main.py with no third-party "
    "dependencies. The program must print its result to stdout when run."
)


def extract_json(text: str) -> dict:
    """Models sometimes wrap JSON in prose or ```json fences. Be forgiving."""
    text = text.strip()
    if text.startswith("```"):
        text = re.sub(r"^```[a-zA-Z]*\n?|\n?```$", "", text).strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        m = re.search(r"\{.*\}", text, re.DOTALL)
        if not m:
            raise
        return json.loads(m.group(0))


def main() -> int:
    task = sys.argv[1] if len(sys.argv) > 1 else "a CLI that prints the first 10 Fibonacci numbers"
    model = sys.argv[2] if len(sys.argv) > 2 else "free-best"

    print(f"📝 Task   : {task}")
    print(f"🤖 Model  : {model}  (routed by OllaBridge Cloud)\n")

    print("→ Asking the router to write the code…")
    raw = chat(task, model=model, system=SYSTEM, temperature=0.1)
    spec = extract_json(raw)

    OUT.mkdir(exist_ok=True)
    written = []
    for f in spec["files"]:
        p = OUT / f["path"]
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(f["content"])
        written.append(p)
        print(f"📄 Wrote  : {p.relative_to(OUT.parent)}  ({len(f['content'])} bytes)")

    run_cmd = spec.get("run", "python main.py")
    print(f"\n▶  Running: {run_cmd}")
    proc = subprocess.run(run_cmd, shell=True, cwd=OUT, capture_output=True, text=True, timeout=60)
    print("─" * 56)
    print(proc.stdout.rstrip() or "(no stdout)")
    if proc.stderr.strip():
        print("[stderr]", proc.stderr.rstrip())
    print("─" * 56)
    print("✅ Done — the router wrote it, your machine ran it." if proc.returncode == 0
          else f"⚠️ exited with code {proc.returncode}")
    return proc.returncode


if __name__ == "__main__":
    raise SystemExit(main())

app2_agent.py

#!/usr/bin/env python3
"""Program 2 — A small *agentic workflow* on the router LLM.

This is the same three-node shape you would wire visually in LangFlow, written
in Python so you can run it and read every step:

    Plan  ->  Research (per sub-question)  ->  Synthesize

Every step is one call to OllaBridge Cloud. Swap the model alias (free-best ->
fable-5 -> local-private) and the workflow is unchanged — that is the whole point
of routing through one gateway.

    python app2_agent.py "How do I start learning to build AI agents?"
"""
from __future__ import annotations

import json
import re
import sys

from ollabridge_client import chat


def plan(question: str, model: str) -> list[str]:
    """Planner agent: break the question into 2-3 focused sub-questions."""
    raw = chat(
        f'Break this into at most 3 focused research sub-questions. '
        f'Return ONLY a JSON array of strings.\n\nQuestion: {question}',
        model=model,
        system="You are a meticulous research planner.",
        temperature=0.2,
    )
    raw = re.sub(r"^```[a-zA-Z]*\n?|\n?```$", "", raw.strip()).strip()
    try:
        subs = json.loads(raw)
        return [str(s) for s in subs][:3] if isinstance(subs, list) else [question]
    except json.JSONDecodeError:
        return [question]


def research(subq: str, model: str) -> str:
    """Researcher agent: answer one sub-question concisely."""
    return chat(
        f"Answer in 2-3 sentences, concrete and practical:\n{subq}",
        model=model,
        system="You are a concise domain expert.",
        temperature=0.3,
    )


def synthesize(question: str, notes: list[tuple[str, str]], model: str) -> str:
    """Writer agent: merge the notes into one clear answer."""
    bundle = "\n\n".join(f"Q: {q}\nA: {a}" for q, a in notes)
    return chat(
        f"Using these notes, write a clear, friendly final answer to the user's "
        f"original question.\n\nORIGINAL: {question}\n\nNOTES:\n{bundle}",
        model=model,
        system="You are a helpful writer. Be encouraging and concrete.",
        temperature=0.4,
    )


def main() -> int:
    question = sys.argv[1] if len(sys.argv) > 1 else "How do I start learning to build AI agents?"
    model = sys.argv[2] if len(sys.argv) > 2 else "free-best"

    print(f"❓ Question: {question}")
    print(f"🤖 Model   : {model}  (routed by OllaBridge Cloud)\n")

    print("🧭 [Planner] decomposing the question…")
    subs = plan(question, model)
    for i, s in enumerate(subs, 1):
        print(f"   {i}. {s}")

    print("\n🔎 [Researcher] answering each sub-question…")
    notes: list[tuple[str, str]] = []
    for i, s in enumerate(subs, 1):
        a = research(s, model)
        notes.append((s, a))
        print(f"   ✓ sub-question {i} answered ({len(a)} chars)")

    print("\n🖊️  [Writer] synthesizing the final answer…\n")
    answer = synthesize(question, notes, model)
    print("=" * 60)
    print(answer)
    print("=" * 60)
    print("\n✅ Agentic workflow complete — 1 plan + "
          f"{len(subs)} research + 1 synthesis calls, all through one router.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

ollabridge_client.py

#!/usr/bin/env python3
"""A tiny, reusable OllaBridge Cloud client.

OllaBridge Cloud is an OpenAI-compatible *router*: one endpoint that picks the
best model behind a logical alias (free-best, fable-5, local-private, ...) with
automatic failover. We talk to it with plain `requests` — no SDK, no surprises,
works behind any firewall.

Configuration (environment variables):
    OLLABRIDGE_URL    e.g. https://app.ollabridge.com   (NO trailing /v1)
    OLLABRIDGE_TOKEN  the device token from pair.py
"""
from __future__ import annotations

import os

import requests

BASE = os.environ.get("OLLABRIDGE_URL", "https://app.ollabridge.com").rstrip("/")
TOKEN = os.environ.get("OLLABRIDGE_TOKEN", "not-needed")


def _headers() -> dict:
    return {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}


def list_models() -> list[str]:
    """Return the model aliases the router can reach right now."""
    r = requests.get(f"{BASE}/v1/models", headers=_headers(), timeout=30)
    r.raise_for_status()
    return [m["id"] for m in r.json().get("data", [])]


def chat(prompt: str, *, model: str = "free-best", system: str | None = None,
         temperature: float = 0.3) -> str:
    """One-shot chat completion through the router. Returns the reply text."""
    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})

    r = requests.post(
        f"{BASE}/v1/chat/completions",
        headers=_headers(),
        json={"model": model, "messages": messages, "temperature": temperature, "stream": False},
        timeout=120,
    )
    r.raise_for_status()
    return (r.json()["choices"][0]["message"]["content"] or "").strip()


if __name__ == "__main__":
    print("Models reachable:")
    for mid in list_models()[:12]:
        print("  -", mid)
    print("\nfree-best says:", chat("Reply with exactly: ROUTER OK", temperature=0))

pair.py

#!/usr/bin/env python3
"""Pair this machine with OllaBridge Cloud and print a device token.

Usage:
    python pair.py ABCD-1234
    python pair.py ABCD-1234 --url https://app.ollabridge.com

Open https://app.ollabridge.com -> Dashboard -> "Pair a device" to get a code,
then run this once. Export the printed token so the demos can use it:

    export OLLABRIDGE_URL=https://app.ollabridge.com
    export OLLABRIDGE_TOKEN=<the token this prints>
"""
from __future__ import annotations

import argparse
import json
import sys
import urllib.request


def pair(code: str, base_url: str) -> dict:
    body = json.dumps({"code": code}).encode("utf-8")
    req = urllib.request.Request(
        f"{base_url.rstrip('/')}/device/pair-simple",
        data=body,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read().decode("utf-8"))


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("code", help="The pairing code shown on the dashboard, e.g. ABCD-1234")
    ap.add_argument("--url", default="https://app.ollabridge.com", help="OllaBridge Cloud base URL")
    args = ap.parse_args()

    result = pair(args.code, args.url)
    if result.get("status") != "ok":
        print(f"Pairing failed: {result.get('error')}", file=sys.stderr)
        return 1

    token = result["device_token"]
    print("Paired! Device:", result.get("device_id"))
    print()
    print("Run these two lines, then run the demos:")
    print(f'  export OLLABRIDGE_URL={args.url}')
    print(f"  export OLLABRIDGE_TOKEN={token}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

requirements.txt

requests>=2.31