api/src/routes/workflows.py - master - ncit/multiagentsystem

api/src/routes/workflows.py
196 строк6 KB
Nikita
feat: initial multi-agent platform scaffold
09 июн 2026, 00:48
09 июн 2026, 00:4819cef10
"""Workflow routes — run, list, create, save, load workflows."""
from __future__ import annotations

import httpx
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

from src.config import settings
from src.workflows.registry import WORKFLOW_TEMPLATES, list_workflow_templates

router = APIRouter()


# ─── Schemas ────────────────────────────────────────────────────────────────

class DAGNodeSchema(BaseModel):
    id: str
    type: str
    agent_type: str | None = None
    prompt_id: str | None = None


class DAGEdgeSchema(BaseModel):
    from_node: str
    to_node: str


class DAGSchema(BaseModel):
    nodes: list[DAGNodeSchema]
    edges: list[DAGEdgeSchema]


class SaveWorkflowRequest(BaseModel):
    name: str
    slug: str | None = None
    description: str = ""
    dag: DAGSchema
    tags: list[str] = []
    is_public: bool = False


class RunWorkflowRequest(BaseModel):
    workflow_id: str
    input_data: dict
    stream: bool = False


# ─── Supabase helper ────────────────────────────────────────────────────────

def _supabase_headers():
    return {
        "apikey": settings.supabase_service_key,
        "Authorization": f"Bearer {settings.supabase_service_key}",
        "Content-Type": "application/json",
        "Prefer": "return=representation",
    }


def _supabase_url(table: str) -> str:
    return f"{settings.supabase_url}/rest/v1/{table}"


# ─── Routes ─────────────────────────────────────────────────────────────────

@router.get("/templates")
async def get_workflow_templates():
    """List all built-in workflow templates."""
    return list_workflow_templates()


@router.post("/save")
async def save_workflow(req: SaveWorkflowRequest, user_id: str = "anonymous"):
    """Save a workflow DAG to Supabase."""
    payload = {
        "slug": req.slug or req.name.lower().replace(" ", "-"),
        "name": req.name,
        "description": req.description,
        "dag": req.dag.model_dump(),
        "tags": req.tags,
        "is_public": req.is_public,
        "owner_id": user_id,
        "version": 1,
    }

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            _supabase_url("workflows"),
            headers=_supabase_headers(),
            json=payload,
        )
        if resp.status_code in (200, 201):
            return resp.json()[0]
        raise HTTPException(resp.status_code, resp.text)


@router.get("/list")
async def list_workflows(user_id: str = "anonymous"):
    """List user's workflows + public workflows."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            _supabase_url("workflows"),
            headers=_supabase_headers(),
            params={
                "or": f"(owner_id.eq.{user_id},is_public.eq.true)",
                "order": "created_at.desc",
            },
        )
        if resp.status_code == 200:
            return resp.json()
        raise HTTPException(resp.status_code, resp.text)


@router.get("/{workflow_id}")
async def get_workflow(workflow_id: str):
    """Get a workflow by ID (from DB or built-in templates)."""
    # Check built-in templates first
    if workflow_id in WORKFLOW_TEMPLATES:
        dag = WORKFLOW_TEMPLATES[workflow_id]
        return {
            "id": workflow_id,
            "name": workflow_id,
            "nodes": [
                {"id": n.id, "type": n.type.value, "agent_type": n.agent_type}
                for n in dag.nodes
            ],
            "edges": [
                {"from": e.from_node, "to": e.to_node}
                for e in dag.edges
            ],
            "is_builtin": True,
        }

    # Query Supabase
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            _supabase_url("workflows"),
            headers=_supabase_headers(),
            params={"id": f"eq.{workflow_id}", "limit": "1"},
        )
        if resp.status_code == 200:
            data = resp.json()
            if data:
                return data[0]
        raise HTTPException(404, f"Workflow '{workflow_id}' not found")


@router.delete("/{workflow_id}")
async def delete_workflow(workflow_id: str, user_id: str = "anonymous"):
    """Delete a user's workflow."""
    async with httpx.AsyncClient() as client:
        resp = await client.delete(
            _supabase_url("workflows"),
            headers=_supabase_headers(),
            params={
                "id": f"eq.{workflow_id}",
                "owner_id": f"eq.{user_id}",
            },
        )
        if resp.status_code == 200:
            return {"deleted": workflow_id}
        raise HTTPException(resp.status_code, resp.text)


@router.post("/run")
async def run_workflow(req: RunWorkflowRequest, user_id: str = "anonymous"):
    """Execute a workflow (dispatched to Trigger.dev in production)."""
    workflow = None

    # Check built-in
    if req.workflow_id in WORKFLOW_TEMPLATES:
        dag = WORKFLOW_TEMPLATES[req.workflow_id]
        workflow = {
            "id": req.workflow_id,
            "dag": {
                "nodes": [
                    {"id": n.id, "type": n.type.value, "agent_type": n.agent_type}
                    for n in dag.nodes
                ],
                "edges": [
                    {"from": e.from_node, "to": e.to_node}
                    for e in dag.edges
                ],
            },
        }

    if not workflow:
        # TODO: load from Supabase
        raise HTTPException(404, f"Workflow '{req.workflow_id}' not found")

    # TODO: dispatch to Trigger.dev for async execution
    return {
        "job_id": f"trigger-wf-{req.workflow_id}-pending",
        "workflow_id": req.workflow_id,
        "status": "queued",
        "dag": workflow["dag"],
    }