diff --git a/src/agent_canvas.html b/src/agent_canvas.html new file mode 100644 index 0000000..f625238 --- /dev/null +++ b/src/agent_canvas.html @@ -0,0 +1,644 @@ + + + + + + Agent Canvas — AgentSandbox + + + + + + + + +
+ + +
+

Agents

+
+
📋 Manager
+
Plans and coordinates the workflow
+
+
+
⚡ Specialist
+
Executes the main task
+
+
+
🔍 Reviewer
+
Audits and validates output
+
+
+
💻 Coder
+
Writes and edits code
+
+
+
🔎 Researcher
+
Searches and synthesizes information
+
+
+
✍️ Writer
+
Drafts and edits text content
+
+ +

Context

+
+
🗺 Planner
+
Breaks down objectives into steps
+
+
+
⚙️ Executor
+
Runs commands and scripts
+
+
+ + +
+
+
+ ← Drag agents from the left panel onto this canvas to build your pipeline → +
+
+
+ + +
+
+

Configure Agent

+ +
+ +
+ + +
+ +
+ + +
+ +
+ + +
+ +
Tools
+
+ + + + +
+
+ +
+ + +
+ Saved: + None yet — save one! +
+ + +
+ + + + diff --git a/src/agents.py b/src/agents.py index 5c1e8d7..e52a6c0 100644 --- a/src/agents.py +++ b/src/agents.py @@ -132,4 +132,39 @@ def build_review_task(reviewer: Agent, plan: str, specialist_output: str) -> Tas ), expected_output="'PASS' or a list of specific issues to fix.", agent=reviewer, - ) \ No newline at end of file + ) + +# ── Canvas-built agents ───────────────────────────────────────────────────────── + +def _all_tools() -> list[BaseTool]: + return [ + WorkspaceWriteTool(), + # Additional tools can be added here + ] + + +def build_agent( + role: str, + goal: str, + backstory: str, + tools: list[BaseTool] | None = None, +) -> Agent: + """Build a CrewAI Agent from canvas configuration.""" + return Agent( + role=role, + goal=goal, + backstory=backstory, + verbose=True, + llm=_llm(), + tools=tools or [], + allow_delegation=False, + ) + + +def build_task_from_prompt(agent: Agent, prompt: str) -> Task: + """Build a CrewAI Task from a freeform prompt string.""" + return Task( + description=prompt, + expected_output="Completed output from this agent.", + agent=agent, + ) diff --git a/src/api.py b/src/api.py index 80f98e0..d092694 100644 --- a/src/api.py +++ b/src/api.py @@ -47,10 +47,9 @@ async def lifespan(app: FastAPI): detail="Job marked failed on startup recovery.", )) yield - # No teardown — db is aiosqlite connection-per-call; graceful shutdown handled by uvicorn -app = FastAPI(title="AgentSandbox", version="0.1.0", lifespan=lifespan) +app = FastAPI(title="AgentSandbox", version="0.2.0", lifespan=lifespan) app.add_middleware( CORSMiddleware, @@ -63,8 +62,22 @@ async def lifespan(app: FastAPI): # ── Request / Response models ─────────────────────────────────────────────────── +class AgentToolConfig(BaseModel): + name: str + enabled: bool = True + + +class AgentPipelineConfig(BaseModel): + type: str + role: str + prompt: str + contextLimit: int = 0 + tools: list[AgentToolConfig] = [] + + class SubmitRequest(BaseModel): objective: str + pipeline: list[AgentPipelineConfig] | None = None # custom pipeline from canvas class JobResponse(BaseModel): @@ -78,12 +91,12 @@ class JobResponse(BaseModel): _active_flow_task: Optional[asyncio.Task] = None -# ── Helpers ─────────────────────────────────────────────────────────────────── +# ── Helpers ──────────────────────────────────────────────────────────────────── -async def _run_flow(job: Job) -> None: +async def _run_flow(job: Job, custom_pipeline: list | None = None) -> None: global _active_flow_task try: - flow = AgentSandboxFlow(db=db, job=job) + flow = AgentSandboxFlow(db=db, job=job, custom_pipeline=custom_pipeline) await flow.run() # Attempt Git commit try: @@ -107,7 +120,7 @@ async def health(): running = await db.get_running_jobs() return { "status": "ok", - "app": "AgentSandbox 0.1.0", + "app": "AgentSandbox 0.2.0", "db": "connected", "workspace": WS_ROOT, "active_jobs": len(running), @@ -123,14 +136,34 @@ async def submit_job(req: SubmitRequest, background: BackgroundTasks): status_code=409, detail="An active job is already running. Wait for it to complete.", ) + job = Job(objective=req.objective) await db.insert_job(job) await db.insert_event(Event( job_id=job.id, event_type="SUBMITTED", detail=f"Objective: {req.objective[:80]}...", )) + + # Serialize pipeline for storage + pipeline_data = None + if req.pipeline: + pipeline_data = [ + { + "type": a.type, + "role": a.role, + "prompt": a.prompt, + "contextLimit": a.contextLimit, + "tools": [{"name": t.name, "enabled": t.enabled} for t in a.tools], + } + for a in req.pipeline + ] + await db.insert_event(Event( + job_id=job.id, event_type="PIPELINE_CONFIGURED", + detail=f"Custom pipeline: {len(req.pipeline)} agents — {[a.role for a in req.pipeline]}", + )) + global _active_flow_task - _active_flow_task = asyncio.create_task(_run_flow(job)) + _active_flow_task = asyncio.create_task(_run_flow(job, custom_pipeline=pipeline_data)) events = await db.get_events(job.id) return JobResponse(job=job, events=events) @@ -169,9 +202,10 @@ async def cancel_job(job_id: str): return job -# ── Dashboard ────────────────────────────────────────────────────────────────── +# ── Static UI routes ─────────────────────────────────────────────────────────── -_dashboard_path = os.path.join(os.path.dirname(__file__), "dashboard.html") +_dashboard_path = os.path.join(os.path.dirname(__file__), "dashboard.html") +_canvas_path = os.path.join(os.path.dirname(__file__), "agent_canvas.html") @app.get("/", response_class=RedirectResponse) @@ -184,11 +218,6 @@ async def dashboard(): return FileResponse(_dashboard_path) -if __name__ == "__main__": - import uvicorn - uvicorn.run( - "src.api:app", - host=os.environ.get("APP_HOST", "127.0.0.1"), - port=int(os.environ.get("APP_PORT", "8090")), - reload=False, - ) +@app.get("/canvas", response_class=HTMLResponse) +async def agent_canvas(): + return FileResponse(_canvas_path) diff --git a/src/flow.py b/src/flow.py index 5624d9c..3dcc460 100644 --- a/src/flow.py +++ b/src/flow.py @@ -1,5 +1,7 @@ """ -AgentSandbox CrewAI Flow — structured orchestration with manager / specialist / reviewer pipeline. +AgentSandbox CrewAI Flow — structured orchestration. +Supports default 3-agent pipeline (manager → specialist → reviewer) +or a custom pipeline from the Agent Canvas. """ from __future__ import annotations @@ -7,9 +9,9 @@ import asyncio import traceback from datetime import datetime, timezone -from typing import Optional +from typing import Any, Optional -from crewai import Crew, Flow +from crewai import Agent as CrewAIAgent, Crew, Flow, Task, LLM from src.db import AgentSandboxDB, Job, JobStatus, Event from src.agents import ( @@ -19,34 +21,40 @@ build_planning_task, build_specialist_task, build_review_task, + build_agent as build_custom_agent, + build_task_from_prompt, + _llm as default_llm, + _all_tools, ) class AgentSandboxFlow(Flow): - """CrewAI Flow that runs manager → specialist → reviewer pipeline for each objective.""" - - def __init__(self, db: AgentSandboxDB, job: Job): + """CrewAI Flow — default 3-agent pipeline or custom pipeline from Agent Canvas.""" + + def __init__( + self, + db: AgentSandboxDB, + job: Job, + custom_pipeline: list[dict] | None = None, + ): super().__init__() self._job = job self._db = db + self._pipeline = custom_pipeline # list of agent configs from canvas - async def _emit(self, event_type: str, detail: Optional[str] = None) -> None: + async def _emit(self, event_type: str, detail: str | None = None) -> None: event = Event(job_id=self._job.id, event_type=event_type, detail=detail) await self._db.insert_event(event) - async def run(self) -> Job: - self._job.status = JobStatus.RUNNING - self._job.started_at = datetime.now(timezone.utc).isoformat() - self._job.updated_at = datetime.now(timezone.utc).isoformat() - await self._db.update_job(self._job) - await self._emit("RUNNING", "Flow started") + # ── Default 3-agent pipeline ──────────────────────────────────────────────── + async def _run_default(self) -> Job: manager = build_manager_agent() specialist = build_specialist_agent() reviewer = build_reviewer_agent() try: - # ── Stage 1: Manager plans ────────────────────────────────────────── + # Stage 1: Manager plans await self._emit("STAGE", "manager_planning") plan_task = build_planning_task(manager, self._job.objective) crew_plan = Crew(agents=[manager], tasks=[plan_task], verbose=True) @@ -58,10 +66,10 @@ async def run(self) -> Job: await self._db.update_job(self._job) await self._emit("MANAGER_PLAN", f"Plan length: {len(plan_text)} chars") - if not plan_text or "error" in plan_text.lower()[:100]: - raise RuntimeError(f"Manager planning failed: {plan_text[:200]}") + if not plan_text: + raise RuntimeError("Manager returned empty plan") - # ── Stage 2: Specialist executes ─────────────────────────────────── + # Stage 2: Specialist executes await self._emit("STAGE", "specialist_execution") specialist_task = build_specialist_task(specialist, plan_text) crew_specialist = Crew(agents=[specialist], tasks=[specialist_task], verbose=True) @@ -73,7 +81,7 @@ async def run(self) -> Job: await self._db.update_job(self._job) await self._emit("SPECIALIST_DONE", f"Output length: {len(specialist_text)} chars") - # ── Stage 3: Reviewer audits ──────────────────────────────────────── + # Stage 3: Reviewer audits await self._emit("STAGE", "reviewer_audit") review_task = build_review_task(reviewer, plan_text, specialist_text) crew_reviewer = Crew(agents=[reviewer], tasks=[review_task], verbose=True) @@ -85,7 +93,6 @@ async def run(self) -> Job: await self._db.update_job(self._job) await self._emit("REVIEWER_DONE", f"Review: {review_text[:200]}") - # ── Finalize ──────────────────────────────────────────────────────── self._job.final_output = ( f"## Plan\n{plan_text}\n\n" f"## Specialist Output\n{specialist_text}\n\n" @@ -98,12 +105,140 @@ async def run(self) -> Job: await self._emit("COMPLETED", "Flow completed successfully") except Exception as exc: - tb = traceback.format_exc() self._job.status = JobStatus.FAILED - self._job.error = f"{type(exc).__name__}: {exc}\n{tb}" + self._job.error = f"{type(exc).__name__}: {exc}\n{traceback.format_exc()}" + self._job.completed_at = datetime.now(timezone.utc).isoformat() + self._job.updated_at = datetime.now(timezone.utc).isoformat() + await self._db.update_job(self._job) + await self._emit("FAILED", str(exc)[:500]) + + return self._job + + # ── Custom pipeline from Agent Canvas ─────────────────────────────────────── + + def _tools_from_config(self, tool_configs: list[dict]) -> list: + """Map canvas tool names → actual crewai_tools tool instances.""" + name_to_tool = {t.name: t for t in _all_tools()} + enabled = [] + for cfg in tool_configs: + name = cfg.get("name") or cfg.get("Name", "") + enabled_flag = cfg.get("enabled", True) + if not enabled_flag: + continue + if name in name_to_tool: + enabled.append(name_to_tool[name]) + return enabled + + def _build_agents_from_pipeline(self, pipeline: list[dict]) -> list[tuple[CrewAIAgent, Task]]: + """Build ordered (agent, task) pairs from canvas pipeline config.""" + agents_and_tasks = [] + prev_result_var = None # Not using crewai memory — chain via task descriptions + + for i, cfg in enumerate(pipeline): + agent_type = cfg.get("type", "").lower() + role = cfg.get("role", f"Agent {i+1}") + prompt = cfg.get("prompt", "") + tool_configs = cfg.get("tools", []) + context_limit = cfg.get("contextLimit", 0) or 0 + + tools = self._tools_from_config(tool_configs) + + # Build the agent + agent = build_custom_agent( + role=role, + goal=prompt, + backstory=prompt, + tools=tools, + ) + + # Build the task — first agent gets the objective, rest get previous output + if i == 0: + task_desc = ( + f"Objective: {self._job.objective}\n\n" + f"Instructions: {prompt}" + ) + else: + task_desc = ( + f"Previous step output:\n{prev_result_var}\n\n" + f"Your instructions: {prompt}" + ) + + task = build_task_from_prompt( + agent=agent, + prompt=task_desc, + ) + + agents_and_tasks.append((agent, task)) + prev_result_var = f"Output from {role} (step {i+1})" + + return agents_and_tasks + + async def _run_custom(self) -> Job: + pipeline = self._pipeline + await self._emit("PIPELINE_START", f"Custom pipeline: {len(pipeline)} agents") + + stages = [] + results = [] + + try: + for i, cfg in enumerate(pipeline): + role = cfg.get("role", f"Agent {i+1}") + await self._emit("STAGE", f"step_{i+1}_{role}") + + agents_and_tasks = self._build_agents_from_pipeline(pipeline[i:]) # rest of pipeline + if not agents_and_tasks: + continue + + (agent, task), *rest = agents_and_tasks + crew = Crew(agents=[agent], tasks=[task], verbose=True) + result = await crew.kickoff_async() + result_text = str(result) if result else "" + + results.append({"agent": role, "result": result_text}) + await self._emit(f"STEP_{i+1}_DONE", f"{role}: {len(result_text)} chars") + + # Store in job fields by slot + slot_map = { + 0: "manager_result", + 1: "specialist_result", + 2: "reviewer_result", + } + field = slot_map.get(i, f"step_{i}_result") + setattr(self._job, field, result_text) + self._job.updated_at = datetime.now(timezone.utc).isoformat() + await self._db.update_job(self._job) + + # Final output = all results concatenated + self._job.final_output = "\n\n".join( + f"## Step {i+1}: {r['agent']}\n{r['result']}" + for i, r in enumerate(results) + ) + self._job.status = JobStatus.COMPLETED + self._job.completed_at = datetime.now(timezone.utc).isoformat() + self._job.updated_at = datetime.now(timezone.utc).isoformat() + await self._db.update_job(self._job) + await self._emit("COMPLETED", f"Custom pipeline completed: {len(results)} steps") + + except Exception as exc: + self._job.status = JobStatus.FAILED + self._job.error = f"{type(exc).__name__}: {exc}\n{traceback.format_exc()}" self._job.completed_at = datetime.now(timezone.utc).isoformat() self._job.updated_at = datetime.now(timezone.utc).isoformat() await self._db.update_job(self._job) await self._emit("FAILED", str(exc)[:500]) return self._job + + # ── Public run ───────────────────────────────────────────────────────────── + + async def run(self) -> Job: + self._job.status = JobStatus.RUNNING + self._job.started_at = datetime.now(timezone.utc).isoformat() + self._job.updated_at = datetime.now(timezone.utc).isoformat() + await self._db.update_job(self._job) + await self._emit("RUNNING", "Flow started") + + if self._pipeline: + return await self._run_custom() + else: + return await self._run_default()