Skip to main content
A flow is a structured sequence of agent-behaviors, tool calls, decision points and interactions designed to achieve a specific goal or complete a multi-step process. Rather than treating each user input or tool invocation as isolated, flows allow the agent (or a team of agents) to follow a predefined or dynamically adapted path, moving through different stages (e.g. information gathering → reasoning or planning → tool use → response) with branching logic and conditional execution as needed. Flows encode higher-level control of behavior: how the agent should proceed, when to invoke memory or knowledge, how to decide among tools, when to ask clarifying questions, and how to wrap up or hand off control.

​
Why Flows are Important

Flows are foundational because they let you move beyond reactive agents and create more reliable, coherent, and maintainable autonomous systems. Some key reasons:
  • Predictability and consistency: With flows, you can ensure that complex tasks are handled in similar ways every time, reducing surprises and unintended behavior.
  • Modularity & reuse: Components of a flow—like gathering context, making decisions, using tools, or summarizing results—can be reused across different flows, making development faster and behavior more consistent.
  • Handling complexity: When tasks are multi-step, involve multiple tools or sources, or need error handling, branching, or feedback loops, flows provide a framework for managing that complexity.
  • Adaptability: Flows can adapt based on what’s observed during execution (e.g. missing information, tool failures). Thus, they allow agents to change course or insert remedial steps without breaking the overall structure.
  • Transparency & debugging: Because flows lay out the intended path of execution, they make it easier for developers to inspect what the agent should do at each stage, diagnose where things might have gone wrong, and improve or test behavior.

​
How the example works:

The flow defines a multi-agent pipeline for triaging, evaluating, and fixing security questions. First, a triage agent classifies an incoming question into one of three categories: “network,” “code,” or “other.” Depending on the category, the flow routes execution to different downstream agents: for “network,” it goes to a network expert that ends the flow; for “code,” it goes to a code evaluator which may either label the code snippet “safe” and terminate, or forward it to a code security fixer for remediation. Between steps, there are conditions (e.g. only send to fixer if vulnerability found) and operations, like EVALUATE, that govern how agents interact and how the flow proceeds. 
from autonomy import Agent, END, Flow, FlowOperation, Model, Node, Repl, START
from sys import argv


async def main(node):
    triage = await Agent.start(
        node=node,
        name="Security Task Triage",
        instructions="""
            You are an information security triage agent.

            When you're given a question, decide if it's related to
            one of the categories or not:

            - network: vulnerabilities in networks, internet, cloud, etc.
            - code: vulnerabilities in code snippets

            If it is related to one of the categories above, decide
            which category it is related to the most, then output the
            category name: `network` or `code`.

            If it is not related to any of the categories
            above, output: `other`.

            Do not say anything else other than network or code, or other.

            Examples:
                Question: is it safe to open all ports on a server?
                Answer: network

                Question: print("Hello world!")
                Answer: code

                Question: why the sky is blue?
                Answer: other
        """,
    )

    network_expert = await Agent.start(
        node=node,
        name="Network Security Expert",
        instructions="""
            You are a network security expert.
            Answer network security questions.
        """,
    )

    code_security_fixer = await Agent.start(
        node=node,
        name="Code Security Fixer",
        instructions="""
            You are a programming expert who knows the Python
            programming language.

            You're given a code snippet and information about a
            vulnerability in it. Change the code snippet precisely to
            fix that vulnerability.

            Don't change anything else.
            Only print the fixed code snippet.
            Under no circumstances print anything else except
            the changed code snippet.
        """,
    )

    code_evaluator = await Agent.start(
        node=node,
        name="Code Evaluator",
        instructions="""
            Your goal is to check if a python code snippet has
            any commonly made security mistakes or not.

            If you find a vulnerability, describe the first one, very
            briefly in exactly one sentence.

            Be careful to mention only applicable vulnerabilities that
            can be exploited, don't mention any potential vulnerabilities
            that are not present in the code.

            Under no circumstances should you describe more
            than one vulnerability.

            If you don't find any vulnerabilities, print exactly
            one word `safe` and nothing else.

            Examples:
                Question:
                    filename = input("Enter file to open: ")
                    with open(f"/safe/dir/{{filename}}") as f:
                        data = f.read()
                Answer: user can exploit filename to open arbitrary files.

                Question:
                    API_KEY = "12345-SECRET-KEY"
                Answer: Hardcoded secrets are not safe.

                Question:
                    import subprocess
                    filename = input("Enter file name to list: ")
                    subprocess.run(["ls", filename])
                Answer: safe

                Question:
                    try:
                        age = int(input("Enter your age: "))
                        if age < 0:
                            raise ValueError("Age can't be negative")
                    except ValueError:
                        print("Invalid age input.")
                Answer: safe

                Question:
                    import ast
                    try:
                        result = ast.literal_eval(user_input)
                    except (ValueError, SyntaxError):
                        print("Invalid expression")
                Answer: safe

                Question:
                    import random
                    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

                    b = bytes([random.randint(0, 255) for _ in range(32)])
                    private_key = Ed25519PrivateKey.from_private_bytes(b)

                    # Sign a message
                    message = b"Sensitive data to sign"
                    signature = private_key.sign(message)

                    # Verify (just to show it's valid with the key pair)
                    public_key = private_key.public_key()
                    public_key.verify(signature, message)

                    print("Signature:", signature.hex())
                    b = bytes([random.randint(0, 255) for _ in range(32)])
                    private_key = Ed25519PrivateKey.from_private_bytes(b)

                    # Sign a message
                    message = b"Sensitive data to sign"
                    signature = private_key.sign(message)

                    # Verify (just to show it's valid with the key pair)
                    public_key = private_key.public_key()
                    public_key.verify(signature, message)

                    print("Signature:", signature.hex())
                Answer: cryptographical keys should be generated only using
                cryptographically secure randomness sources.

                Question:
                    import sqlite3

                    conn = sqlite3.connect('example.db')
                    cursor = conn.cursor()

                    cursor.execute("INSERT INTO users (username, password) VALUES (?, ?)", (username, password))
                    conn.commit()

                    # Clean up
                    cursor.close()
                    conn.close()
                Answer: safe
            """,
    )

    flow = Flow()

    flow.add(START, triage)

    flow.add(triage, network_expert, condition="network")
    flow.add(triage, code_evaluator, condition="code", operation=FlowOperation.EVALUATE)

    flow.add(network_expert, END)

    flow.add(code_evaluator, code_security_fixer)
    flow.add(code_evaluator, END, condition="safe")
    flow.add(code_security_fixer, code_evaluator, operation=FlowOperation.EVALUATE)

    flow = await Flow.start(node, flow)

    await Repl.start(flow, "localhost:7000")


Node.start(main)
images/main/Dockerfile
FROM ghcr.io/build-trust/autonomy-python
COPY . .
ENTRYPOINT ["python", "main.py"]
autonomy.yaml
name: example008
pods:
  - name: main-pod
    public: true
    containers:
      - name: main
        image: main
⌘I