GitHub - memgrafter/flatmachines: Orchestrate agents with a peer network of state machines. Supports checkpoint/restore, persistence, and more. BYO cli, or agent, or use flatagents with API. Batteries included.

# 1. Set an API key e.g.
export OPENAI_API_KEY="sk-..."

# 2. Run the example (sets up venv, installs deps, runs)
cd sdk/examples/helloworld/python
./run.sh

pip install flatmachines[flatagents] flatagents[litellm]

spec: flatprofile
spec_version: "4.1.0"
data:
  model_profiles:
    fast:
      provider: openai
      name: gpt-5-mini
      max_tokens: 2048

    quality:
      provider: openai
      name: gpt-5
      max_tokens: 4096

  # Used when a flatagent has no model field
  default: fast

  # Uncomment to force every agent to this profile
  # override: quality

spec: flatagent
spec_version: "4.1.0"
data:
  name: hello-world-agent
  model:
    profile: fast
  system: >
    You are an agent in a test-time sequential scaling.
    Reply with exactly one output character in text format.
    No explanation. No wrapper.
  user: |
    Target: {{ input.target }}
    Built so far: {{ input.current }}
    Next character:

spec: flatmachine
spec_version: "4.1.0"
data:
  name: hello-world-loop
  context:
    target: "{{ input.target }}"
    current: ""
  agents:
    builder: ./agent.yml
  hooks: hello-world-hooks
  states:
    start:
      type: initial
      transitions:
        - condition: "context.current == context.target"
          to: done
        - to: build_char
    build_char:
      agent: builder
      execution:
        type: retry
        retry_on_empty: true
        backoffs: [2, 8, 16, 35]
        jitter: 0.1
      input:
        current: "{{ context.current }}"
        target: "{{ context.target }}"
      output_to_context:
        expected_char: "{{ context.target[context.current|length] }}"
        last_output: "{{ output.next_char or output.content }}"
      transitions:
        - condition: "context.last_output != null and context.last_output != '' and context.last_output[0] == context.expected_char"
          to: append_char
        - to: build_char
    append_char:
      action: append_char
      transitions:
        - condition: "context.current == context.target"
          to: done
        - to: build_char
    done:
      type: final
      output:
        result: "{{ context.current }}"
        success: true

import asyncio
from flatmachines import FlatMachine, HooksRegistry, LoggingHooks, setup_logging, get_logger

setup_logging(level="INFO")
logger = get_logger(__name__)

class HelloWorldHooks(LoggingHooks):
    def on_action(self, action_name, context):
        if action_name == "append_char":
            last_output = context.get("last_output", "")
            if last_output:
                context["current"] = context["current"] + str(last_output)[0]
        return context

async def main():
    hooks_registry = HooksRegistry()
    hooks_registry.register("hello-world-hooks", HelloWorldHooks)

    machine = FlatMachine(
        config_file="config/machine.yml",
        hooks_registry=hooks_registry,
    )

    result = await machine.execute(input={"target": "Hello, World!"}, max_agent_calls=20)
    logger.info(f"Result: '{result.get('result', '')}'")
    logger.info(f"Success: {result.get('success')}")

asyncio.run(main())

from flatmachines import setup_logging
setup_logging(level="INFO")