Build an Ultracode-Like Coding Workflow with Agno and AgentOS

TL;DR

• This design uses Agno-only orchestration. Workflow is the scheduler, native Agent instances are the reasoning roles, and AgentOS is the runtime/API surface. External Git and sandbox infrastructure is still required.
• Use one custom workflow step for dynamic worker fan-out. Agno’s Parallel block is excellent when branches are known while defining the workflow; a planner-generated worker count is easier to execute with asyncio.gather() inside a custom step.
• Store compact workflow metadata in Agno session state, but keep large diffs, logs, and test artifacts in object storage.
• Do not let model output determine what gets merged. A workspace manager must derive changed files from Git, compose one candidate, run deterministic tests, and promote only the candidate envelope reviewed through workflow HITL.
• Use a bounded Loop for implementation and repair rounds. Stop after a small fixed number of rounds instead of letting an Agno Team task loop run indefinitely.
• AgentOS can persist sessions, paused workflow state, traces, and workflow HITL requirements in PostgreSQL. A workflow confirmation is not the separate AgentOS /approvals record created by an @approval-decorated tool.
• For production, add a sandbox worker pool, lease manager, immutable artifact store, short-lived credentials, tenant authorization, budgets, and cleanup.

What You Will Learn Here

• How to map the Ultracode pattern onto Agno primitives
• When to use Workflow, Loop, Parallel, and custom function steps
• How to create typed planner, worker, reducer, and reviewer agents
• How to run a dynamic set of coding workers concurrently
• How to integrate and test one immutable candidate before review
• How to pause before merge and resume through AgentOS
• Which infrastructure Agno supplies and which infrastructure you still own
• How to evaluate the Agno workflow against one coding agent

This is Part 2 of the Ultracode Architecture series. Part 1 derives the portable architecture. Part 3 is an independent LangGraph alternative. This article assumes you chose Agno and implements the coding-workflow pattern without LangGraph or another external workflow engine.

Why Agno Can Approximate the Pattern

Agno now exposes three relevant layers:

Agno Teams also support coordinate, route, broadcast, and task modes. I would not use a Team as the top-level scheduler here.

The workflow already owns:

• round count
• worker fan-out
• candidate integration
• review gates
• repair termination
• merge review

Adding a Team leader that independently creates and updates another task list would give the system two schedulers. Use a Workflow for explicit repair-round control. TeamMode.tasks is also bounded by max_iterations, which defaults to 10, but its leader owns a separate model-directed task loop.

The architecture is:

AgentOS API
    |
    v
Agno Workflow
    |
    +--> Plan step --------------------> planner Agent
    |
    +--> Bounded Loop
    |      |
    |      +--> prepare round
    |      +--> dynamic asyncio fan-out
    |      |      +--> coding Agent in workspace A
    |      |      +--> coding Agent in workspace B
    |      |      +--> coding Agent in workspace C
    |      +--> deterministic candidate integration
    |      +--> deterministic test runner
    |      +--> reducer Agent
    |      +--> verifier + refuter Agents
    |      +--> stop when passed or exhausted
    |
    +--> top-level candidate output review
    |
    +--> finalize
           +--> passed + changed: promote exact reviewed SHA
           +--> passed + read-only: return report
           +--> failed: return failure outcome

Project Structure

agno-ultracode/
  app/
    schemas.py
    agents.py
    workspace.py
    workflow.py
    runtime.py
  evals/
    cases.py
  tests/
    test_plan_invariants.py
    test_candidate_binding.py
    test_resume.py
  pyproject.toml

The examples were checked on July 15, 2026 against Python 3.10+ and Agno 2.7.3.

uv add "agno[os,openai,postgres]==2.7.3" \
  "psycopg[binary,pool]" sqlalchemy pydantic

Step 1: Define Contracts the Model Cannot Rewrite

# app/schemas.py
from typing import Literal

from pydantic import BaseModel, Field, model_validator


class CodingRequest(BaseModel):
    repository: str
    base_sha: str
    request: str
    success_criteria: list[str] = Field(min_length=1)


class WorkItem(BaseModel):
    id: str
    objective: str
    read_only: bool
    write_globs: list[str] = Field(default_factory=list)
    additional_suite_ids: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_permissions(self):
        if self.read_only and self.write_globs:
            raise ValueError("read-only work cannot declare writes")
        if not self.read_only and not self.write_globs:
            raise ValueError("writable work requires declared scopes")
        return self


class WorkflowPlan(BaseModel):
    substantive: bool
    reason: str
    success_criteria: list[str] = Field(min_length=1)
    tasks: list[WorkItem] = Field(min_length=1, max_length=4)

    @model_validator(mode="after")
    def validate_wave(self):
        expected = range(2, 5) if self.substantive else range(1, 2)
        if len(self.tasks) not in expected:
            raise ValueError("routine plans need 1 task; substantive plans need 2-4")
        ids = [task.id for task in self.tasks]
        if len(ids) != len(set(ids)):
            raise ValueError("task IDs must be unique")
        return self


class WorkerOutput(BaseModel):
    status: Literal["completed", "blocked"]
    summary: str
    unresolved_risks: list[str] = Field(default_factory=list)


class ReviewOutput(BaseModel):
    candidate_id: str
    candidate_sha: str
    test_report_digest: str
    evidence_digest: str
    candidate_passed: bool
    blocking_issues: list[str] = Field(default_factory=list)
    evidence: list[str] = Field(default_factory=list)

    @model_validator(mode="after")
    def validate_review(self):
        if self.candidate_passed and self.blocking_issues:
            raise ValueError("passing reviews cannot contain blocking issues")
        return self


class SynthesisOutput(BaseModel):
    candidate_id: str
    candidate_sha: str
    test_report_digest: str
    evidence_digest: str
    summary: str
    acceptance_evidence: list[str] = Field(default_factory=list)
    unresolved_risks: list[str] = Field(default_factory=list)

The planner may suggest scopes. The workspace service must expand them against the repository and reject concrete overlap before workers start.

Step 2: Create Narrow Native Agents

# app/agents.py
import os

from agno.agent import Agent
from agno.models.openai import OpenAIResponses

from app.schemas import ReviewOutput, SynthesisOutput, WorkerOutput, WorkflowPlan


def model() -> OpenAIResponses:
    return OpenAIResponses(
        id=os.environ["AGENT_MODEL_ID"],
        retries=2,
        delay_between_retries=1,
        exponential_backoff=True,
    )


def planner(tools: list) -> Agent:
    return Agent(
        id="ultracode-planner",
        model=model(),
        tools=tools,
        output_schema=WorkflowPlan,
        instructions=[
            "Inspect the repository before planning.",
            "Create one task for routine work or 2-4 independent tasks.",
            "Declare write scopes and optional additional verification suite IDs.",
            "Do not weaken the supplied success criteria during repair.",
        ],
    )


def worker(task_id: str, tools: list) -> Agent:
    return Agent(
        id=f"ultracode-worker-{task_id}",
        model=model(),
        tools=tools,
        output_schema=WorkerOutput,
        instructions=[
            "Complete only the assigned task.",
            "Use only the provided workspace tools.",
            "Report blocked work honestly.",
        ],
    )


def reviewer(role: str) -> Agent:
    instruction = (
        "Verify the candidate against every success criterion."
        if role == "verifier"
        else "Try to falsify the candidate using a concrete counterexample."
    )
    return Agent(
        id=f"ultracode-{role}",
        model=model(),
        output_schema=ReviewOutput,
        instructions=[
            instruction,
            "Use immutable diff and test evidence.",
            "candidate_passed means the exact candidate is safe to approve.",
        ],
    )


def reducer() -> Agent:
    return Agent(
        id="ultracode-reducer",
        model=model(),
        output_schema=SynthesisOutput,
        instructions=[
            "Summarize the immutable candidate and deterministic test evidence.",
            "Do not hide failed tests, contradictions, or unresolved risks.",
        ],
    )

Use stable Agno session_id values for correlation. Add an Agno database to the agents only if you want their individual sessions persisted; the workflow session remains the orchestration source of truth.

Step 3: Define the Workspace Boundary

Agno’s Workspace and CodingTools toolkits scope operations to a root directory and can require confirmation. Their documentation is explicit that path scoping is not process isolation.

The workflow should depend on a service like this:

# app/workspace.py
from typing import Protocol

from pydantic import BaseModel

from app.schemas import CodingRequest, WorkItem, WorkerOutput


class WorkerRecord(BaseModel):
    task_id: str
    round: int
    workspace_id: str
    base_sha: str
    changed_files: list[str]
    artifact_uris: list[str]
    output: WorkerOutput


class Candidate(BaseModel):
    id: str
    sha: str
    base_sha: str
    original_base_sha: str
    round: int
    changed_files: list[str]  # Cumulative from original_base_sha.
    diff_uri: str  # Full diff from original_base_sha to sha.


class TestReport(BaseModel):
    candidate_id: str
    candidate_sha: str
    passed: bool
    exit_codes: dict[str, int]
    artifact_uri: str
    digest: str


class EvidenceManifest(BaseModel):
    candidate_id: str
    candidate_sha: str
    diff_uri: str
    test_artifact_uri: str
    test_report_digest: str
    diff_text: str
    test_log_text: str
    digest: str


class WorkspaceService(Protocol):
    async def planner_tools(self, request: CodingRequest, *, user_id: str) -> list: ...
    async def resolve_suite_ids(
        self,
        request: CodingRequest,
        *,
        user_id: str,
        proposed_suite_ids: list[str],
    ) -> list[str]:
        """Return required plus allowlisted optional suites from server policy."""
        ...
    async def create_worker(
        self,
        request: CodingRequest,
        task: WorkItem,
        *,
        user_id: str,
        round: int,
        idempotency_key: str,
        parent_candidate_id: str | None,
    ) -> tuple[str, list]: ...
    async def inspect_and_freeze(
        self,
        workspace_id: str,
        task: WorkItem,
        output: WorkerOutput,
        *,
        round: int,
    ) -> WorkerRecord: ...
    async def compose(
        self,
        records: list[WorkerRecord],
        *,
        parent_candidate_id: str | None,
    ) -> Candidate: ...
    async def test(self, candidate: Candidate, suite_ids: list[str]) -> TestReport: ...
    async def evidence(
        self,
        candidate: Candidate,
        report: TestReport,
    ) -> EvidenceManifest: ...
    async def cancel(self, workspace_id: str) -> None: ...
    async def promote(
        self,
        candidate: Candidate,
        *,
        expected_sha: str,
        idempotency_key: str,
    ) -> dict: ...

This service—not an agent—must:

• derive changed files from Git
• reject writes outside declared globs
• resolve required and optional suite IDs through repository/tenant policy
• compose patches into one candidate workspace
• preserve original_base_sha and derive cumulative changed files/full diff from that base on every repair round
• detect merge conflicts
• run commands and capture real exit codes
• hash immutable diff and test artifacts
• promote only the approved candidate hash

For trusted local development, the service can use Git worktrees. For untrusted repositories, run Agno tools inside a container, VM, Daytona workspace, or equivalent sandbox.

Step 4: Implement One Dynamic Round

Agno Parallel is ideal for a known set of branches. Here the planner decides the worker count at runtime, so one custom async step performs the fan-out.

# app/workflow.py
import asyncio
import json

from agno.run import RunContext
from agno.workflow import Loop, Step, StepInput, StepOutput, Workflow
from agno.workflow.types import OnError, OnReject
from pydantic import BaseModel

from app.agents import planner, reducer, reviewer, worker
from app.schemas import (
    CodingRequest,
    ReviewOutput,
    SynthesisOutput,
    WorkflowPlan,
    WorkerOutput,
)
from app.workspace import (
    Candidate,
    EvidenceManifest,
    TestReport,
    WorkerRecord,
    WorkspaceService,
)


class CandidateEnvelope(BaseModel):
    passed: bool
    candidate_id: str
    candidate_sha: str
    original_base_sha: str
    changed_files: list[str]
    test_report_digest: str
    evidence_digest: str
    diff_uri: str
    test_artifact_uri: str
    test_exit_codes: dict[str, int]
    synthesis: dict


def build_workflow(db, workspaces: WorkspaceService):
    async def create_plan(
        step_input: StepInput,
        run_context: RunContext,
    ) -> StepOutput:
        request = CodingRequest.model_validate(step_input.input)
        user_id = run_context.user_id
        if not user_id:
            raise PermissionError("authenticated user_id is required")
        tools = await workspaces.planner_tools(request, user_id=user_id)
        baseline_suite_ids = await workspaces.resolve_suite_ids(
            request,
            user_id=user_id,
            proposed_suite_ids=[],
        )
        if not baseline_suite_ids:
            raise RuntimeError("repository policy returned no required suites")
        response = await planner(tools).arun(
            {
                "request": request.request,
                "immutable_success_criteria": request.success_criteria,
                "required_suite_ids": baseline_suite_ids,
            },
            user_id=user_id,
            session_id=f"{run_context.session_id}:planner",
        )
        plan = WorkflowPlan.model_validate(response.content).model_copy(
            update={"success_criteria": request.success_criteria}
        )
        run_context.session_state.update(
            {
                "request": request.model_dump(),
                "user_id": user_id,
                "success_criteria": request.success_criteria,
                "baseline_suite_ids": baseline_suite_ids,
                "plan": plan.model_dump(),
                "round": 0,
                "parent_candidate_id": None,
                "round_passed": False,
                "exhausted": False,
            }
        )
        return StepOutput(content=plan.model_dump_json())

    async def execute_round(
        step_input: StepInput,
        run_context: RunContext,
    ) -> StepOutput:
        state = run_context.session_state
        request = CodingRequest.model_validate(state["request"])
        plan = WorkflowPlan.model_validate(state["plan"])
        round_number = state["round"]
        user_id = state["user_id"]

        async def run_worker(task):
            workspace_id = None
            try:
                workspace_id, tools = await workspaces.create_worker(
                    request,
                    task,
                    user_id=user_id,
                    round=round_number,
                    idempotency_key=(
                        f"{run_context.session_id}:{round_number}:{task.id}:workspace"
                    ),
                    parent_candidate_id=state["parent_candidate_id"],
                )
                response = await worker(task.id, tools).arun(
                    {
                        "objective": task.objective,
                        "workspace_id": workspace_id,
                        "additional_suite_ids": task.additional_suite_ids,
                    },
                    user_id=user_id,
                    session_id=(
                        f"{run_context.session_id}:worker:{round_number}:{task.id}"
                    ),
                )
                output = WorkerOutput.model_validate(response.content)
                return await workspaces.inspect_and_freeze(
                    workspace_id,
                    task,
                    output,
                    round=round_number,
                )
            except asyncio.CancelledError:
                if workspace_id:
                    await asyncio.shield(workspaces.cancel(workspace_id))
                raise

        child_tasks = [
            asyncio.create_task(run_worker(task))
            for task in plan.tasks
        ]
        try:
            records = await asyncio.gather(*child_tasks)
        except BaseException:
            for child in child_tasks:
                child.cancel()
            await asyncio.gather(*child_tasks, return_exceptions=True)
            raise
        if any(record.output.status != "completed" for record in records):
            raise RuntimeError("a worker did not complete")

        candidate = await workspaces.compose(
            records,
            parent_candidate_id=state["parent_candidate_id"],
        )
        expected_base_sha = (
            request.base_sha
            if round_number == 0
            else Candidate.model_validate(state["candidate"]).sha
        )
        if (
            candidate.base_sha != expected_base_sha
            or candidate.original_base_sha != request.base_sha
            or any(record.base_sha != expected_base_sha for record in records)
        ):
            raise RuntimeError("candidate lineage does not match expected base")
        suite_ids = await workspaces.resolve_suite_ids(
            request,
            user_id=user_id,
            proposed_suite_ids=[
                suite_id
                for task in plan.tasks
                for suite_id in task.additional_suite_ids
            ],
        )
        if not set(state["baseline_suite_ids"]).issubset(suite_ids):
            raise RuntimeError("verification policy dropped a required suite")
        report = await workspaces.test(candidate, suite_ids)
        if (
            report.candidate_id != candidate.id
            or report.candidate_sha != candidate.sha
        ):
            raise RuntimeError("test report does not match candidate")

        evidence = EvidenceManifest.model_validate(
            await workspaces.evidence(candidate, report)
        )
        if (
            evidence.candidate_id != candidate.id
            or evidence.candidate_sha != candidate.sha
            or evidence.test_report_digest != report.digest
        ):
            raise RuntimeError("evidence is not bound to candidate and tests")
        synthesis_response = await reducer().arun(
            {
                "success_criteria": state["success_criteria"],
                "candidate": candidate.model_dump(),
                "test_report": report.model_dump(),
                "evidence": evidence.model_dump(),
                "worker_records": [record.model_dump() for record in records],
            },
            user_id=user_id,
            session_id=f"{run_context.session_id}:reducer:{round_number}",
        )
        synthesis = SynthesisOutput.model_validate(synthesis_response.content)
        if (
            synthesis.candidate_id != candidate.id
            or synthesis.candidate_sha != candidate.sha
            or synthesis.test_report_digest != report.digest
            or synthesis.evidence_digest != evidence.digest
        ):
            raise RuntimeError("synthesis is not bound to immutable evidence")
        review_payload = json.dumps(
            {
                "success_criteria": state["success_criteria"],
                "candidate": candidate.model_dump(),
                "test_report": report.model_dump(),
                "evidence": evidence.model_dump(),
                "synthesis": synthesis.model_dump(),
            }
        )
        verifier_result, refuter_result = await asyncio.gather(
            reviewer("verifier").arun(
                review_payload,
                user_id=user_id,
                session_id=f"{run_context.session_id}:verifier:{round_number}",
            ),
            reviewer("refuter").arun(
                review_payload,
                user_id=user_id,
                session_id=f"{run_context.session_id}:refuter:{round_number}",
            ),
        )
        verifier = ReviewOutput.model_validate(verifier_result.content)
        refuter = ReviewOutput.model_validate(refuter_result.content)
        for review in (verifier, refuter):
            if (
                review.candidate_id != candidate.id
                or review.candidate_sha != candidate.sha
                or review.test_report_digest != report.digest
                or review.evidence_digest != evidence.digest
            ):
                raise RuntimeError("review is not bound to immutable evidence")

        passed = (
            report.passed
            and verifier.candidate_passed
            and refuter.candidate_passed
        )
        state.update(
            {
                "candidate": candidate.model_dump(),
                "test_report": report.model_dump(),
                "reviews": {
                    "verifier": verifier.model_dump(),
                    "refuter": refuter.model_dump(),
                },
                "synthesis": synthesis.model_dump(),
                "evidence": evidence.model_dump(),
                "round_passed": passed,
            }
        )

        if not passed:
            if round_number >= 2:
                state["exhausted"] = True
            else:
                repair_response = await planner(
                    await workspaces.planner_tools(request, user_id=user_id)
                ).arun(
                    json.dumps(
                        {
                            "request": request.request,
                            "immutable_success_criteria": state["success_criteria"],
                            "candidate": candidate.model_dump(),
                            "test_report": report.model_dump(),
                            "evidence": evidence.model_dump(),
                            "synthesis": synthesis.model_dump(),
                            "reviews": state["reviews"],
                            "instruction": "Create one bounded repair wave.",
                        }
                    ),
                    user_id=user_id,
                    session_id=(
                        f"{run_context.session_id}:repair:{round_number + 1}"
                    ),
                )
                repair_plan = WorkflowPlan.model_validate(repair_response.content)
                state["plan"] = repair_plan.model_copy(
                    update={"success_criteria": state["success_criteria"]}
                ).model_dump()
                state["parent_candidate_id"] = candidate.id
                state["round"] = round_number + 1

        return StepOutput(
            content=json.dumps(
                {
                    "round": round_number,
                    "passed": passed,
                    "exhausted": state["exhausted"],
                    "candidate_sha": candidate.sha,
                }
            ),
            success=passed or not state["exhausted"],
        )

    def end_loop(outputs: list[StepOutput]) -> bool:
        result = json.loads(outputs[-1].content)
        return result["passed"] or result["exhausted"]

    def candidate_envelope(
        step_input: StepInput,
        run_context: RunContext,
    ) -> StepOutput:
        state = run_context.session_state
        candidate = Candidate.model_validate(state["candidate"])
        envelope = CandidateEnvelope(
            passed=state["round_passed"],
            candidate_id=candidate.id,
            candidate_sha=candidate.sha,
            original_base_sha=candidate.original_base_sha,
            changed_files=candidate.changed_files,
            test_report_digest=state["test_report"]["digest"],
            evidence_digest=state["evidence"]["digest"],
            diff_uri=state["evidence"]["diff_uri"],
            test_artifact_uri=state["evidence"]["test_artifact_uri"],
            test_exit_codes=state["test_report"]["exit_codes"],
            synthesis=state["synthesis"],
        )
        return StepOutput(content=envelope.model_dump_json())

    async def finalize(
        step_input: StepInput,
        run_context: RunContext,
    ) -> StepOutput:
        state = run_context.session_state
        envelope = CandidateEnvelope.model_validate_json(
            step_input.previous_step_content
        )
        candidate = Candidate.model_validate(state["candidate"])
        if (
            envelope.candidate_id != candidate.id
            or envelope.candidate_sha != candidate.sha
            or envelope.original_base_sha != candidate.original_base_sha
            or envelope.test_report_digest != state["test_report"]["digest"]
            or envelope.evidence_digest != state["evidence"]["digest"]
        ):
            raise RuntimeError("reviewed envelope does not match session candidate")
        if not envelope.passed:
            return StepOutput(
                content=json.dumps(
                    {
                        "status": "failed",
                        "round": state["round"],
                        "reviews": state.get("reviews"),
                    }
                )
            )
        if candidate.sha == candidate.original_base_sha:
            return StepOutput(
                content=json.dumps(
                    {
                        "status": "completed",
                        "kind": "report",
                        "candidate_sha": candidate.sha,
                        "synthesis": state["synthesis"],
                    }
                )
            )
        result = await workspaces.promote(
            candidate,
            expected_sha=envelope.candidate_sha,
            idempotency_key=(
                f"{run_context.session_id}:merge:{candidate.sha}"
            ),
        )
        return StepOutput(content=json.dumps(result))

    review_candidate_step = Step(
        name="Review Candidate Envelope",
        executor=candidate_envelope,
        requires_output_review=lambda output: (
            CandidateEnvelope.model_validate_json(output.content).passed
            and CandidateEnvelope.model_validate_json(output.content).candidate_sha
            != CandidateEnvelope.model_validate_json(output.content).original_base_sha
        ),
        output_review_message=(
            "Review the candidate SHA, deterministic tests, and evidence."
        ),
        on_reject=OnReject.cancel,
        max_retries=0,
        on_error=OnError.fail,
    )

    return Workflow(
        id="agno-ultracode",
        name="Agno Ultracode Approximation",
        db=db,
        input_schema=CodingRequest,
        session_state={},
        store_events=True,
        steps=[
            Step(
                name="Plan",
                executor=create_plan,
                max_retries=0,
                on_error=OnError.fail,
            ),
            Loop(
                name="Implementation and Review",
                steps=[
                    Step(
                        name="Execute Round",
                        executor=execute_round,
                        max_retries=0,
                        on_error=OnError.fail,
                    )
                ],
                end_condition=end_loop,
                max_iterations=3,
            ),
            review_candidate_step,
            Step(
                name="Finalize",
                executor=finalize,
                max_retries=0,
                on_error=OnError.fail,
            ),
        ],
    )

This is intentionally one scheduler:

• Workflow controls order.
• Loop controls maximum rounds.
• the custom step controls dynamic workers.
• agents only make typed reasoning decisions.
• the workspace service controls candidate truth.

Important Corrections for Production

The code is a scaffold, not a drop-in sandbox platform.

Before deployment:

1. Validate that planner tasks have unique IDs and non-overlapping concrete files.
2. Make workspace creation idempotent by session, round, and task.
3. Quarantine or reset workspaces on timeout and cancellation.
4. Derive changed files and test exit codes outside the model.
5. Keep candidate review as a top-level workflow step; nested Condition steps do not enforce pre-execution confirmation in the same way.
6. Include candidate ID, SHA, test digest, and evidence digest in the reviewed envelope, then compare them again before promotion.
7. Ensure repair plans preserve the original baseline commands and success criteria.
8. Clean workspaces on success, failure, cancellation, and TTL expiry.

Agno workflow state persists at workflow boundaries when a database is configured. A custom dynamic fan-out step that crashes may need to rerun, so every external operation still needs an idempotency key. This design uses persisted workflow output review, not AgentOS’s separate @approval tool-record system.

Step 5: Serve It with AgentOS

# app/runtime.py
import os

from agno.db.postgres import PostgresDb
from agno.os import AgentOS
from agno.os.config import AuthorizationConfig

from app.bootstrap import production_workspaces
from app.workflow import build_workflow


workflow_db = PostgresDb(
    db_url=os.environ["AGNO_DATABASE_URL"],
    session_table="ultracode_workflow_sessions",
)
agentos_db = PostgresDb(
    db_url=os.environ["AGENTOS_DATABASE_URL"],
)

workflow = build_workflow(workflow_db, production_workspaces)

agent_os = AgentOS(
    id="ultracode-agent-os",
    workflows=[workflow],
    db=agentos_db,
    tracing=True,
    authorization=True,
    authorization_config=AuthorizationConfig(
        verification_keys=[os.environ["JWT_VERIFICATION_KEY"]],
        algorithm="RS256",
        user_isolation=True,
    ),
)

app = agent_os.get_app()

Use a postgresql+psycopg://... URI for Agno PostgresDb.

AgentOS provides workflow endpoints, stored runs, SSE-compatible streaming, cancellation APIs, HITL continuation, and trace visibility. In this example, JWT authorization and user isolation are both enabled; authorization alone does not scope every database read by user.

Version-specific warning: in the Agno 2.7.3 tag, the paused workflow SSE response path loads a session by session_id without passing the scoped user_id. Do not expose that paused-stream path to untrusted multi-tenant callers on this pinned version without an ownership-enforcing gateway or patch. Upgrade after verifying the fix, or disable the affected stream and use authenticated run/status endpoints that enforce ownership. This caution comes from the tagged workflow_response_streamer and user_scope middleware, not from a general claim that every AgentOS route is unscoped.

Start, continue, and cancel

run = await workflow.arun(
    input=CodingRequest(
        repository="github.com/acme/api",
        base_sha="abc123",
        request="Audit every route for missing authorization checks.",
        success_criteria=["Every route has an explicit authorization decision."],
    ),
    session_id="ultra-123",
    user_id="user-42",
)

if run.is_paused:
    # Trusted in-process example; production uses authenticated HITL handling.
    for requirement in run.steps_requiring_output_review:
        requirement.confirm()
    run = await workflow.acontinue_run(run_response=run)

In production, resolve workflow HITL through authenticated AgentOS continue routes or an admin UI that verifies run ownership. The direct library calls above bypass AgentOS authentication and are suitable only for trusted in-process administration.

Important limitation: workflow.acancel_run() marks the workflow for cooperative cancellation, but manually invoked Agents inside this custom dynamic step are not automatically registered as native child Step executors. A production implementation needs a cancellation bridge that tracks child run IDs and workspace IDs, calls each Agent’s cancellation API, terminates sandbox process groups, and awaits all child tasks.

Infrastructure Requirements

Client / IDE / CI
       |
       v
AgentOS API + JWT authorization
       |
       v
Agno Workflow runtime
       |
       +--> workflow Postgres
       +--> trace Postgres / OTel exporter
       +--> persisted workflow HITL state
       +--> lease / concurrency controller
       +--> object storage
       |
       v
Sandbox worker pool
  |        |        |
Agno     Agno     Agno
worker   worker   worker
  |        |        |
isolated repository workspaces
       |
       v
Git provider + CI

PostgreSQL is enough for a small queue and lease table. Add Redis, Valkey, or a managed queue only when cross-replica signaling and higher concurrency justify another operational dependency.

Security Boundaries

Agno gives useful controls, but use them at the right layer:

• Workspace scopes file tools to a root and confirmation policy.
• CodingTools can restrict commands and base directories when configured.
• AgentOS deep-copies mutable runtime objects per request.
• JWT authorization is opt-in.
• User-level data isolation is opt-in.

None of those controls turns arbitrary repository shell execution into a process sandbox.

For multi-tenant production:

• use ephemeral containers or stronger isolation
• deny network by default
• issue short-lived credentials per worker
• mount only the assigned repository and artifact paths
• cap CPU, memory, disk, time, output, and tokens
• terminate the process group on cancellation
• treat repository files as untrusted instructions

Failure Semantics

Agno cancellation is cooperative. Your workspace provider must implement the hard process stop.

Expected test or review rejection is a domain outcome and may still end with a technically completed workflow run. Clients must inspect the final content (promoted, report, failed, or cancelled) rather than treating run status alone as business success.

Evaluation

Compare:

1. One Agno coding Agent
2. Agno Team in task mode
3. This Agno Workflow

Use the same model, repository commit, tools, tasks, and budget.

Test at least:

• routine one-worker tasks
• parallel read-only audits
• disjoint multi-file changes
• overlapping-write plans that must be rejected
• failed tests followed by repair
• candidate review rejection
• process restart and resume
• cancellation during a worker command

Agno eval suites can combine Agent-as-Judge and Reliability checks and gate CI with an exit code. Use those for agent behavior, then add deterministic workflow assertions for candidate hashes, tool calls, HITL review, and side-effect duplication.

What This Approximation Does Not Copy

It does not reproduce:

• Claude Code’s private trigger heuristic
• generated JavaScript workflows
• Claude Code’s context assembly
• provider-specific xhigh effort
• its workflow UI or internal convergence policy

It does recreate the useful engineering pattern:

typed plan
+ bounded parallel workers
+ persistent session state
+ deterministic candidate and tests
+ role-conditioned verification
+ human review
+ resumable runtime

Final Recommendation

Start with:

• one Agno Workflow
• native Agno role agents
• PostgreSQL
• a trusted-repository worktree adapter
• no more than four concurrent workers
• two repair rounds
• mandatory candidate/test/review evidence before merge

Move to container or VM sandboxes before untrusted or multi-tenant use.

The operating rule is:

Agno Workflow schedules. Agents reason. Workspace infrastructure proves. AgentOS persists and governs.

Related Reading

• Claude Code Ultracode Architecture Blueprint for Cursor and Custom Agents
• Build an Ultracode-Like Coding Workflow with LangGraph
• Agno/AgentOS + FastAPI vs LangChain/LangGraph + FastAPI
• 12-Factor Agents in Practice: LangChain/LangGraph and Agno/AgentOS
• Secure Agentic Code Generation

Sources

• Agno Docs, “Building Workflows”, accessed July 15, 2026.
• Agno Docs, “Advanced Workflow Patterns”, accessed July 15, 2026.
• Agno Docs, “Parallel Workflow”, accessed July 15, 2026.
• Agno Docs, “Loop Steps reference”, accessed July 15, 2026.
• Agno Docs, “Custom Function Step Workflow”, accessed July 15, 2026.
• Agno Docs, “Workflow Session State”, accessed July 15, 2026.
• Agno Docs, “Workflow Input Schema”, accessed July 15, 2026.
• Agno Docs, “Human-in-the-Loop in Workflows”, accessed July 15, 2026.
• Agno Docs, “Workflow Output Review”, accessed July 15, 2026.
• Agno Docs, “Approval”, accessed July 15, 2026.
• Agno Docs, “Workflow reference”, accessed July 15, 2026.
• Agno Docs, “Workspace toolkit”, accessed July 15, 2026.
• Agno Docs, “CodingTools”, accessed July 15, 2026.
• Agno Docs, “Security and Auth”, accessed July 15, 2026.
• Agno Docs, “Per-User Data Isolation”, accessed July 15, 2026.
• Agno Docs, “Background Execution”, accessed July 15, 2026.
• Agno Docs, “Agent Observability”, accessed July 15, 2026.
• Agno Docs, “Evals overview”, accessed July 15, 2026.
• PyPI, agno 2.7.3, July 14, 2026.