Skip to content

Build workflows

A workflow is an async Python function that runs agent roles on a repository. Start one with agl new.

Create a workflow

agl new implement

This creates ~/.agl/workspace/workflows/implement/ with two files.

__init__.py:

from agl.sdk import Run, workflow

@workflow
async def implement(run: Run) -> None:
    ...

pyproject.toml:

[project]
name = "implement"
version = "0.1.1"

[project.entry-points."agl.workflows"]
implement = "implement:implement"

[tool.agl]
requires = "agents-gl>=0.1.0"
config = []

The name is also the module AGL imports, so it is a Python identifier, such as implement or fix_issues, and not the name of a standard-library module.

Open ~/.agl/workspace in your editor, not the workflow's directory, so the editor finds AGL.

What a workflow needs

  • A directory in ~/.agl/workspace/workflows/.
  • A pyproject.toml with a name, a version and an entry under agl.workflows. The entry's name is what agl run takes.
  • An __init__.py.
  • An async function decorated with @workflow whose first parameter is annotated as a Run.

requires sets the oldest AGL the workflow runs on, and agl new writes the version you have. config lists the repository settings it reads. Check that it loads with agl workflows implement.

A small workflow

This workflow does what you ask and commits the result. Its three files go in the directory agl new implement made, beside its pyproject.toml.

roles.py:

from dataclasses import dataclass
from agl.sdk import (
    Claude,
    ClaudeEffort,
    Restriction,
    Role,
    describe,
    prompt_file,
    reporting_tool,
    role,
)

@dataclass(frozen=True)
class Summary:
    text: str = describe("What you changed.")

@role(model=Claude.OPUS(effort=ClaudeEffort.HIGH), accepts=(str,))
def builder_role() -> Role[Summary]:
    return Role(
        name="builder",
        instructions=prompt_file("prompts/builder.md"),
        restrictions={Restriction.NO_NETWORK},
        tools=[reporting_tool("report", "Say what you changed.", Summary)],
    )

builder = builder_role()

prompts/builder.md:

Do this in the repository: {{str}}
When you are done, report what you changed.

__init__.py:

from dataclasses import dataclass
from agl.sdk import Run, arg, workflow
from .roles import builder

@dataclass(frozen=True)
class Params:
    request: str = arg("-r", "--request", help="What to do.")

@workflow
async def implement(run: Run[Params]) -> None:
    await run.step(builder, run.params.request, commit="Do what the run was asked")

Run it:

agl run implement -n first-try -r "Add a health check endpoint"

When it finishes, its work is on the branch agl/first-try.

Reference

agl.sdk.workflow(fn=None)

Declares an async function as a workflow agl run can start.

Parameters:

  • fn (_Function[P] | None, default: None ) –

    The async def below a bare @workflow. Nothing there means @workflow() was written with parentheses, which is refused.

Returns:

Raises:

  • InputError –

    @workflow was written with parentheses, or fn isn't an async def.

agl.sdk.Workflow dataclass

A workflow declared with @workflow.