Real-World Use Cases

1. Automated Code Review in CI/CD

Run the LYDOS Project Analyzer from GitHub Actions on every pull request. The job fails the build if the agent reports a non-success status.

GitHub Actions workflow

# .github/workflows/lydos-review.yml
name: LYDOS Code Review

on:
  pull_request:
    types: [opened, synchronize, reopened]

jobs:
  review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install lydos-sdk
        run: pip install ./sdk-py

      - name: Run LYDOS Project Analyzer on the diff
        env:
          LYDOS_BASE_URL: ${{ secrets.LYDOS_BASE_URL }}
          LYDOS_API_KEY:  ${{ secrets.LYDOS_API_KEY }}
        run: python3 scripts/lydos_pr_review.py

scripts/lydos_pr_review.py

#!/usr/bin/env python3
"""Run LYDOS Project Analyzer against the current checkout and fail the build on non-success."""
import asyncio
import os
import sys

from lydos import LydosClient
from lydos.exceptions import LydosError


async def main() -> int:
    async with LydosClient(
        base_url=os.environ["LYDOS_BASE_URL"],
        api_key=os.environ["LYDOS_API_KEY"],
        timeout=120.0,
    ) as client:
        try:
            # SPEC-001 = LYDOS Project Analyzer (5 Quality Gates, Groq-powered analysis)
            result = await client.run_agent(
                "SPEC-001",
                {
                    "task": "Review PR diff for security and correctness",
                    "path": "./",
                    "depth": "standard",
                },
            )
        except LydosError as exc:
            print(f"::error::LYDOS agent failed: {exc}")
            return 2

        print(f"::notice::LYDOS Project Analyzer finished in {result.execution_time_ms} ms")
        if result.output:
            for key, value in result.output.items():
                print(f"  {key}: {value}")

        return 0 if result.status == "completed" else 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))

Expected outcome

• Build fails when LYDOS Project Analyzer reports status != "completed"
• Full agent output surfaces in the Actions log as annotations
• The script exits 2 on transport/auth errors and 1 on agent failures

2. Research Pipeline with the Deep Research Agent

Fan out research requests to the deep_research Python agent using submit_task + Q63 async polling, then synthesize the findings through chat().

#!/usr/bin/env python3
"""Weekly research digest — Q63 async task submission + polling."""
import asyncio
import os
from datetime import datetime

from lydos import LydosClient
from lydos.exceptions import LydosError

TOPICS = [
    "Latest developments in AI agent frameworks",
    "Emerging cloud security threats",
    "Machine learning compiler optimisations",
]


async def research(client: LydosClient, topic: str) -> dict:
    task = await client.submit_task(
        agent_type="deep_research",
        payload={"query": topic, "depth": "comprehensive"},
    )

    while True:
        current = await client.get_task(task.task_id)
        if current.status == "completed":
            return {"topic": topic, "result": current.result or {}}
        if current.status == "failed":
            return {"topic": topic, "error": current.error}
        await asyncio.sleep(2.0)


async def main() -> None:
    async with LydosClient(
        base_url=os.environ.get("LYDOS_BASE_URL", "http://localhost:8888"),
        api_key=os.environ.get("LYDOS_API_KEY"),
        timeout=300.0,
    ) as client:
        # 1. Run all topics in parallel
        research_results = await asyncio.gather(
            *(research(client, topic) for topic in TOPICS),
            return_exceptions=True,
        )

        findings = [r for r in research_results if isinstance(r, dict) and "result" in r]
        print(f"Collected {len(findings)} research topics")

        # 2. Synthesize a digest using the chat endpoint
        try:
            digest = await client.chat(
                message=(
                    "Write a concise weekly digest in Markdown "
                    f"from these research findings: {findings}"
                ),
                provider="auto",
                max_tokens=1500,
                temperature=0.4,
            )
        except LydosError as exc:
            print(f"synthesis failed: {exc}")
            return

        out_path = f"digest_{datetime.utcnow():%Y%m%d}.md"
        with open(out_path, "w", encoding="utf-8") as fh:
            fh.write(digest.text)
        print(f"wrote {out_path} — {digest.total_tokens} tokens")


if __name__ == "__main__":
    asyncio.run(main())

Expected outcome

• Three deep research tasks run concurrently via Q63
• Failures on individual topics don't abort the digest
• Final Markdown is written to disk with the token count logged

3. Health-driven remediation

Poll the real health() endpoint, detect degraded modules, and hand them to the system_architect agent for a remediation plan. No WebSocket or streaming required — the SDK exposes the same REST surface the dashboard uses.

#!/usr/bin/env python3
"""Watch LYDOS health and run a remediation agent when modules degrade."""
import asyncio
import os

from lydos import LydosClient
from lydos.exceptions import LydosError

POLL_SECONDS = 30
DEGRADED_THRESHOLD = 70


async def remediate(client: LydosClient, degraded: list[dict]) -> None:
    result = await client.run_agent(
        "system_architect",
        {
            "task": "Diagnose degraded LYDOS modules and propose fixes",
            "modules": degraded,
        },
    )
    print(f"remediation status={result.status} latency={result.execution_time_ms}ms")
    if result.output:
        print("recommendations:", result.output.get("recommendations"))


async def main() -> None:
    async with LydosClient(
        base_url=os.environ.get("LYDOS_BASE_URL", "http://localhost:8888"),
        api_key=os.environ.get("LYDOS_API_KEY"),
    ) as client:
        while True:
            try:
                health = await client.health()
            except LydosError as exc:
                print(f"health poll failed: {exc}")
                await asyncio.sleep(POLL_SECONDS)
                continue

            print(f"overall_score={health.overall_score} status={health.overall_status}")

            degraded = [
                {"name": m.name, "category": m.category, "score": m.score, "status": m.status}
                for m in health.modules
                if m.score is not None and m.score < DEGRADED_THRESHOLD
            ]

            if degraded:
                print(f"!! {len(degraded)} degraded modules — running remediation agent")
                await remediate(client, degraded)

            await asyncio.sleep(POLL_SECONDS)


if __name__ == "__main__":
    asyncio.run(main())

4. Capability dashboard widget (TypeScript)

Build a small React widget that pulls live capability stats from the TypeScript SDK. This uses the real getCapabilityStats() and queryCapabilities() methods — no fake @lydos/sdk package, no fake streaming API.

// app/components/CapabilityWidget.tsx
"use client";

import { useEffect, useState } from "react";
import { LydosClient, LydosClientError, type Capability, type CapabilityStats } from "@lydos/client";

const client = new LydosClient({
  baseURL: process.env.NEXT_PUBLIC_LYDOS_BASE_URL ?? "http://localhost:8888",
  apiKey: process.env.NEXT_PUBLIC_LYDOS_API_KEY,
  timeout: 10_000,
  retry: {
    maxRetries: 2,
    baseDelayMs: 300,
    maxDelayMs: 2_000,
    retryableStatuses: [429, 500, 502, 503, 504],
    jitter: true,
  },
});

export function CapabilityWidget() {
  const [stats, setStats] = useState<CapabilityStats | null>(null);
  const [missing, setMissing] = useState<Capability[]>([]);
  const [error, setError] = useState<string | null>(null);

  useEffect(() => {
    let cancelled = false;

    async function load() {
      try {
        const [nextStats, nextMissing] = await Promise.all([
          client.getCapabilityStats(),
          client.queryCapabilities({ status: "missing", limit: 5 }),
        ]);
        if (cancelled) return;
        setStats(nextStats);
        setMissing(nextMissing);
      } catch (err) {
        if (cancelled) return;
        if (err instanceof LydosClientError) {
          setError(`HTTP ${err.statusCode ?? "n/a"}: ${err.message}`);
        } else {
          setError(String(err));
        }
      }
    }

    load();
    return () => {
      cancelled = true;
    };
  }, []);

  if (error) return <div className="text-red-500">LYDOS: {error}</div>;
  if (!stats) return <div>Loading capability stats…</div>;

  return (
    <section>
      <h2>Capability coverage</h2>
      <p>
        {stats.implemented}/{stats.total} implemented · {stats.tested} tested · {stats.deployed} deployed
      </p>

      <h3>Top missing</h3>
      <ul>
        {missing.map((cap) => (
          <li key={cap.id}>
            <strong>{cap.name}</strong> — {cap.category} ({cap.risk_level})
          </li>
        ))}
      </ul>
    </section>
  );
}

5. Memory-backed assistant

Combine store_memory(), search_memory(), and chat() to build a minimal RAG assistant against the real /api/memory/* endpoints.

#!/usr/bin/env python3
"""Tiny RAG loop: search semantic memory, then answer with chat()."""
import asyncio
import os

from lydos import LydosClient


async def main() -> None:
    async with LydosClient(
        base_url=os.environ.get("LYDOS_BASE_URL", "http://localhost:8888"),
        api_key=os.environ.get("LYDOS_API_KEY"),
    ) as client:
        # Seed memory with a few facts (idempotent — upsert semantics server-side).
        for fact in [
            {"content": "LYDOS ships agents across 12 departments", "metadata": {"topic": "overview"}},
            {"content": "Q63 provides user-scoped async tasks via submit_task/get_task", "metadata": {"topic": "api"}},
            {"content": "The Python SDK package name is lydos-sdk (not @lydos/sdk)", "metadata": {"topic": "sdk"}},
        ]:
            await client.store_memory(content=fact["content"], metadata=fact["metadata"])

        question = "How many agents does LYDOS ship with?"
        hits = await client.search_memory(query=question, limit=3)

        context = "\n".join(f"- {hit.content}" for hit in hits)
        reply = await client.chat(
            message=(
                "Use the context below to answer the question. "
                f"Context:\n{context}\n\nQuestion: {question}"
            ),
            provider="auto",
            max_tokens=256,
            temperature=0.2,
        )
        print(reply.text)


if __name__ == "__main__":
    asyncio.run(main())

Patterns to reuse

from lydos.exceptions import (
    AuthenticationError,
    RateLimitError,
    ServerError,
    TimeoutError,
    ValidationError,
    LydosError,
)

try:
    result = await client.run_agent("SPEC-001", {"task": "audit"})  # SPEC-001 = LYDOS Project Analyzer
except AuthenticationError as exc:
    print(f"401/403: {exc.detail} (status={exc.status_code})")
except RateLimitError as exc:
    print(f"429: retry after {exc.retry_after}s")
except (ServerError, TimeoutError) as exc:
    print(f"server issue: {exc}")
except ValidationError as exc:
    print(f"400/422: {exc.detail}")
except LydosError as exc:
    print(f"unknown SDK failure: {exc}")

import { LydosClientError } from "@lydos/client";

try {
  // SPEC-001 = LYDOS Project Analyzer (maps to /api/agents/SPEC-001/run)
  const result = await client.runAgent("SPEC-001", { target: "./src", async: false });
  console.log(result.status, result.output);
} catch (err) {
  if (err instanceof LydosClientError) {
    switch (err.statusCode) {
      case 401:
      case 403:
        console.error("auth failed:", err.response);
        break;
      case 429:
        console.error("rate limited, check Retry-After header");
        break;
      default:
        console.error(`HTTP ${err.statusCode ?? "n/a"}: ${err.message}`);
    }
  } else {
    throw err;
  }
}

async def batch(client, agent_type, items):
    tasks = [
        await client.submit_task(agent_type=agent_type, payload={"input": item})
        for item in items
    ]
    async def wait(task):
        while True:
            current = await client.get_task(task.task_id)
            if current.status in {"completed", "failed"}:
                return current
            await asyncio.sleep(1)
    return await asyncio.gather(*(wait(t) for t in tasks))