File size: 3,934 Bytes
a10e62e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """
Composition API Routes - Multi-skill workflow execution.
Endpoints:
- POST /composition/execute - Execute skill composition workflow
- POST /composition/validate - Validate workflow DAG
- GET /composition/status/{id} - Get workflow execution status
Reference: Phase 60 Plan 03
"""
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from typing import List, Optional, Dict, Any
from core.database import get_db
from core.skill_composition_engine import SkillCompositionEngine, SkillStep
router = APIRouter(prefix="/composition", tags=["composition"])
class StepModel(BaseModel):
step_id: str = Field(..., description="Unique step identifier")
skill_id: str = Field(..., description="Skill ID to execute")
inputs: Dict[str, Any] = Field(default_factory=dict, description="Input parameters")
dependencies: List[str] = Field(default_factory=list, description="Step IDs this depends on")
condition: Optional[str] = Field(None, description="Conditional execution")
timeout_seconds: int = Field(30, ge=1, le=300, description="Step timeout")
class WorkflowRequest(BaseModel):
workflow_id: str = Field(..., description="Unique workflow identifier")
agent_id: str = Field(..., description="Agent ID executing workflow")
steps: List[StepModel] = Field(..., min_items=1, description="Workflow steps")
class WorkflowResponse(BaseModel):
success: bool
workflow_id: str
execution_id: Optional[str] = None
results: Optional[Dict[str, Any]] = None
error: Optional[str] = None
duration_seconds: Optional[float] = None
@router.post("/execute", response_model=WorkflowResponse)
async def execute_workflow(
request: WorkflowRequest,
db: Session = Depends(get_db)
):
"""Execute a skill composition workflow."""
engine = SkillCompositionEngine(db)
# Convert to SkillStep objects
steps = [
SkillStep(
step_id=s.step_id,
skill_id=s.skill_id,
inputs=s.inputs,
dependencies=s.dependencies,
condition=s.condition,
timeout_seconds=s.timeout_seconds
)
for s in request.steps
]
result = await engine.execute_workflow(
workflow_id=request.workflow_id,
steps=steps,
agent_id=request.agent_id
)
return result
@router.post("/validate")
def validate_workflow(
request: WorkflowRequest,
db: Session = Depends(get_db)
):
"""Validate workflow DAG without executing."""
engine = SkillCompositionEngine(db)
steps = [
SkillStep(
step_id=s.step_id,
skill_id=s.skill_id,
inputs=s.inputs,
dependencies=s.dependencies
)
for s in request.steps
]
result = engine.validate_workflow(steps)
return result
@router.get("/status/{execution_id}")
def get_workflow_status(
execution_id: str,
db: Session = Depends(get_db)
):
"""Get workflow execution status."""
from core.models import SkillCompositionExecution
workflow = db.query(SkillCompositionExecution).filter(
SkillCompositionExecution.id == execution_id
).first()
if not workflow:
raise HTTPException(status_code=404, detail="Workflow execution not found")
return {
"execution_id": workflow.id,
"workflow_id": workflow.workflow_id,
"status": workflow.status,
"validation_status": workflow.validation_status,
"current_step": workflow.current_step,
"completed_steps": workflow.completed_steps or [],
"rollback_performed": workflow.rollback_performed,
"started_at": workflow.started_at.isoformat(),
"completed_at": workflow.completed_at.isoformat() if workflow.completed_at else None,
"duration_seconds": workflow.duration_seconds,
"error": workflow.error_message
}
|