Build workflows¶
A workflow is an async Python function that runs agent roles on a repository.
Start one with agl new.
Create a workflow¶
This creates ~/.agl/workspace/workflows/implement/ with two files.
__init__.py:
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.tomlwith aname, aversionand an entry underagl.workflows. The entry's name is whatagl runtakes. - An
__init__.py. - An async function decorated with
@workflowwhose first parameter is annotated as aRun.
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:
__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:
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 defbelow a bare@workflow. Nothing there means@workflow()was written with parentheses, which is refused.
Returns:
Raises:
-
InputError–@workflowwas written with parentheses, orfnisn't anasync def.
agl.sdk.Workflow
dataclass
¶
A workflow declared with @workflow.