### Programmatic Client Setup with OpenAI Adapter
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Configure the client programmatically for full control, explicitly constructing and registering provider adapters. This example shows setting up an OpenAI adapter with custom configurations.
```python
adapter = OpenAIAdapter(
api_key = "sk-வுகளை...",
base_url = "https://custom-endpoint.example.com/v1",
default_headers = { "X-Custom": "value" },
timeout = 30.0
)
client = Client(
providers = { "openai": adapter },
default_provider = "openai"
)
```
--------------------------------
### Simple Linear Workflow Example
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
A basic example of a linear workflow with start, end, and intermediate steps. Defines graph-level goals and layout direction.
```dot
digraph Simple {
graph [goal="Run tests and report"]
rankdir=LR
start [shape=Mdiamond, label="Start"]
exit [shape=Msquare, label="Exit"]
run_tests [label="Run Tests", prompt="Run the test suite and report results"]
report [label="Report", prompt="Summarize the test results"]
start -> run_tests -> report -> exit
}
```
--------------------------------
### Environment Context Block Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
This block provides structured runtime information to the agent. It is generated at session start and included in every system prompt.
```text
Working directory: {working_directory}
Is git repository: {true/false}
Git branch: {current_branch}
Platform: {darwin/linux/windows}
OS version: {os_version_string}
Today's date: {YYYY-MM-DD}
Model: {model_display_name}
Knowledge cutoff: {knowledge_cutoff_date}
```
--------------------------------
### LoggingExecutionEnvironment Wrapper Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Shows how to wrap an ExecutionEnvironment to add logging for command execution.
```plaintext
-- Logging wrapper
LoggingExecutionEnvironment(inner: ExecutionEnvironment):
exec_command(cmd, ...):
LOG("exec: " + cmd)
result = inner.exec_command(cmd, ...)
LOG("exit: " + result.exit_code + " in " + result.duration_ms + "ms")
RETURN result
```
--------------------------------
### KubernetesExecutionEnvironment File Read Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Shows how to read a file from a Kubernetes pod using `kubectl cp`.
```plaintext
-- File operations use kubectl cp
read_file(path) -> kubectl cp : /dev/stdout
```
--------------------------------
### Client Setup from Environment Variables
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Use this method for most applications to automatically configure the client by reading environment variables for registered providers.
```python
client = Client.from_env()
```
--------------------------------
### KubernetesExecutionEnvironment Command Execution Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Illustrates executing commands within a Kubernetes pod using `kubectl exec`.
```plaintext
-- Commands execute in a Kubernetes pod
exec_command(cmd, ...) -> kubectl exec -- sh -c
```
--------------------------------
### Rate Limiting Middleware Example
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Provides an example of a middleware function for proactive rate limiting using a token bucket. This function ensures requests are made only when the budget is available.
```text
FUNCTION rate_limit_middleware(request, next):
token_bucket.acquire() -- block until budget available
RETURN next(request)
```
--------------------------------
### Example LLM Model Catalog
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
An example implementation of the model catalog, listing various models from Anthropic, OpenAI, and Gemini with their respective details. This catalog is advisory and should be kept updated.
```plaintext
MODELS = [
-- ==========================================================
-- Anthropic -- prefer Claude Opus 4.6 for top quality
-- ==========================================================
ModelInfo(id="claude-opus-4-6", provider="anthropic", display_name="Claude Opus 4.6", context_window=200000, supports_tools=true, supports_vision=true, supports_reasoning=true),
ModelInfo(id="claude-sonnet-4-5", provider="anthropic", display_name="Claude Sonnet 4.5", context_window=200000, supports_tools=true, supports_vision=true, supports_reasoning=true),
-- ==========================================================
-- OpenAI -- prefer the latest GPT-5+ model for top quality
-- ==========================================================
ModelInfo(id="gpt-5.2", provider="openai", display_name="GPT-5.2", context_window=1047576, supports_tools=true, supports_vision=true, supports_reasoning=true),
ModelInfo(id="gpt-5.2-mini", provider="openai", display_name="GPT-5.2 Mini", context_window=1047576, supports_tools=true, supports_vision=true, supports_reasoning=true),
ModelInfo(id="gpt-5.2-codex", provider="openai", display_name="GPT-5.2 Codex", context_window=1047576, supports_tools=true, supports_vision=true, supports_reasoning=true),
-- ==========================================================
-- Gemini -- prefer Gemini 3.1 Pro Preview for top quality
-- ==========================================================
ModelInfo(id="gemini-3.1-pro-preview", provider="gemini", display_name="Gemini 3.1 Pro Preview", context_window=1048576, supports_tools=true, supports_vision=true, supports_reasoning=true),
ModelInfo(id="gemini-3-flash-preview", provider="gemini", display_name="Gemini 3 Flash (Preview)", context_window=1048576, supports_tools=true, supports_vision=true, supports_reasoning=true),
]
```
--------------------------------
### DockerExecutionEnvironment File Read Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Shows how to read a file from a Docker container using `docker cp`.
```plaintext
-- File operations use volume mounts or docker cp
read_file(path) -> docker cp : - | read
```
--------------------------------
### DockerExecutionEnvironment File Write Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Demonstrates writing content to a file within a Docker container using `docker cp`.
```plaintext
write_file(path, content) -> pipe content | docker cp - :
```
--------------------------------
### RemoteSSHExecutionEnvironment Command Execution Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Demonstrates executing commands on a remote host over SSH.
```plaintext
-- Commands execute over SSH
exec_command(cmd, ...) -> ssh
```
--------------------------------
### Example Request with Provider Options
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Illustrates how to use the `provider_options` field to pass provider-specific parameters, such as Anthropic's thinking configuration and beta headers.
```python
request = Request(
model = "claude-opus-4-6",
messages = [ ... ],
provider_options = {
"anthropic": {
"thinking": { "type": "enabled", "budget_tokens": 10000 },
"beta_headers": ["interleaved-thinking-2025-05-14"]
}
}
)
```
--------------------------------
### Start Handler Implementation
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
A no-operation handler for the pipeline's entry point that immediately returns a SUCCESS outcome.
```Go
StartHandler:
FUNCTION execute(node, context, graph, logs_root) -> Outcome:
RETURN Outcome(status=SUCCESS)
```
--------------------------------
### DockerExecutionEnvironment Command Execution Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Illustrates how commands are executed within a Docker container using `docker exec`.
```plaintext
-- Commands execute inside a Docker container
exec_command(cmd, ...) -> docker exec sh -c
```
--------------------------------
### Tool Execute Handler Example
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
An example of a synchronous tool execute handler that takes parsed arguments and returns a string result. The handler can define default values for its parameters.
```python
FUNCTION get_weather(location: String, unit: String = "celsius") -> String:
-- Call weather API...
RETURN "72F and sunny in " + location
```
--------------------------------
### RemoteSSHExecutionEnvironment File Read Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Illustrates reading a file from a remote host using SFTP.
```plaintext
-- File operations use SCP/SFTP
read_file(path) -> sftp get :
```
--------------------------------
### Human Gate Workflow Example
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Demonstrates a workflow incorporating a human-in-the-loop gate using the hexagon shape. Shows conditional branching based on human input.
```dot
digraph Review {
rankdir=LR
start [shape=Mdiamond, label="Start"]
exit [shape=Msquare, label="Exit"]
review_gate [
shape=hexagon,
label="Review Changes",
type="wait.human"
]
start -> review_gate
review_gate -> ship_it [label="[A] Approve"]
review_gate -> fixes [label="[F] Fix"]
ship_it -> exit
fixes -> review_gate
}
```
--------------------------------
### ReadOnlyExecutionEnvironment Wrapper Example
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Demonstrates wrapping an ExecutionEnvironment to disable write operations.
```plaintext
-- Read-only wrapper (rejects all writes)
ReadOnlyExecutionEnvironment(inner: ExecutionEnvironment):
write_file(path, content):
RAISE "Write operations are disabled in read-only mode"
exec_command(cmd, ...):
-- Could analyze command for write intent, or allow all
RETURN inner.exec_command(cmd, ...)
```
--------------------------------
### DOT Graph Example with Stylesheet
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Defines a pipeline graph using DOT syntax, including a graph-level stylesheet to specify LLM models and providers for different node classes.
```dot
digraph Pipeline {
graph [
goal="Implement feature X",
model_stylesheet="
* { llm_model: claude-sonnet-4-5; llm_provider: anthropic; }
.code { llm_model: claude-opus-4-6; llm_provider: anthropic; }
#critical_review { llm_model: gpt-5.2; llm_provider: openai; reasoning_effort: high; }
"
]
start [shape=Mdiamond]
exit [shape=Msquare]
plan [label="Plan", class="planning"]
implement [label="Implement", class="code"]
critical_review [label="Critical Review", class="code"]
start -> plan -> implement -> critical_review -> exit
}
```
--------------------------------
### Condition Expression Examples
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Illustrates various ways to use condition expressions for routing in the Attractor system, including success, failure, context flags, and specific values.
```pseudocode
-- Route on success
plan -> implement [condition="outcome=success"]
-- Route on failure
plan -> fix [condition="outcome=fail"]
-- Route on success AND a context flag
validate -> deploy [condition="outcome=success && context.tests_passed=true"]
-- Route when a context value is not a specific value
review -> iterate [condition="context.loop_state!=exhausted"]
-- Route based on preferred label
gate -> fix [condition="preferred_label=Fix"]
```
--------------------------------
### Logging Middleware Example
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
A middleware function that logs request details and response token usage. It wraps provider calls to observe and log information.
```pseudocode
FUNCTION logging_middleware(request, next):
LOG("Request to " + request.provider + "/" + request.model)
response = next(request)
LOG("Response: " + response.usage.total_tokens + " tokens")
RETURN response
client = Client(
providers = { ... },
middleware = [logging_middleware]
)
```
--------------------------------
### Example Delays with Default RetryPolicy
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Shows the calculated delay for each retry attempt using the default RetryPolicy settings (base=1.0, multiplier=2.0, max=60.0), including the approximate range with jitter.
```text
| Attempt | Base Delay | With Jitter (approx range) |
|---------|------------|---------------------------|
| 0 | 1.0s | 0.5s -- 1.5s |
| 1 | 2.0s | 1.0s -- 3.0s |
| 2 | 4.0s | 2.0s -- 6.0s |
| 3 | 8.0s | 4.0s -- 12.0s |
| 4 | 16.0s | 8.0s -- 24.0s |
```
--------------------------------
### Streaming Middleware Example
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
A middleware function designed to handle streaming requests by wrapping the event iterator. It allows observation and transformation of individual stream events.
```pseudocode
FUNCTION streaming_middleware(request, next):
event_iterator = next(request)
FOR EACH event IN event_iterator:
log_event(event)
YIELD event
```
--------------------------------
### Execute Single Tool Call
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Executes a single tool call, including looking up the tool, validating arguments, and emitting events for the start and end of the call.
```pseudocode
FUNCTION execute_single_tool(session, tool_call):
session.emit(TOOL_CALL_START, tool_name = tool_call.name, call_id = tool_call.id)
-- Look up tool in registry
registered = session.provider_profile.tool_registry.get(tool_call.name)
IF registered IS None:
error_msg = "Unknown tool: " + tool_call.name
session.emit(TOOL_CALL_END, call_id = tool_call.id, error = error_msg)
RETURN ToolResult(tool_call_id = tool_call.id, content = error_msg, is_error = true)
-- Validate arguments against the tool schema
IF NOT validate_arguments(tool_call.arguments, registered.definition.parameters):
```
--------------------------------
### Setting Default Client
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Demonstrates how to set a module-level default client or pass a client explicitly for individual calls.
```pseudocode
set_default_client(my_client)
-- Or pass explicitly per call:
result = generate(model = "...", prompt = "...", client = my_client)
```
--------------------------------
### Session State Transitions
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Illustrates the valid transitions between different Session states during operation.
```go
IDLE -> PROCESSING -- on submit()
PROCESSING -> PROCESSING -- tool loop continues
PROCESSING -> AWAITING_INPUT -- model asks user a question (no tool calls, open-ended)
PROCESSING -> IDLE -- natural completion or turn limit
PROCESSING -> CLOSED -- unrecoverable error
IDLE -> CLOSED -- explicit close()
any -> CLOSED -- abort signal (after graceful shutdown cleanup)
AWAITING_INPUT -> PROCESSING -- user provides answer
```
--------------------------------
### Get Pipeline Status
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Retrieves the current status and progress of a running or completed pipeline using its unique identifier.
```APIDOC
## GET /pipelines/{id}
### Description
Get pipeline status and progress.
### Method
GET
### Endpoint
/pipelines/{id}
#### Path Parameters
- **id** (string) - Required - The unique identifier of the pipeline.
### Response
#### Success Response (200)
- **status** (string) - The current status of the pipeline (e.g., running, completed, failed).
- **progress** (object) - Details about the pipeline's progress.
```
--------------------------------
### Run Directory Structure
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Illustrates the standard directory layout created for each pipeline execution, including logs, checkpoints, and artifacts.
```text
{logs_root}/
checkpoint.json -- Serialized checkpoint after each node
manifest.json -- Pipeline metadata (name, goal, start time)
{node_id}/
status.json -- Node execution outcome
prompt.md -- Rendered prompt sent to LLM
response.md -- LLM response text
artifacts/
{artifact_id}.json -- File-backed artifacts
```
--------------------------------
### Get Checkpoint State
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Retrieves the current checkpoint state of a pipeline. Checkpoints allow pipelines to be resumed from a specific point.
```APIDOC
## GET /pipelines/{id}/checkpoint
### Description
Get current checkpoint state.
### Method
GET
### Endpoint
/pipelines/{id}/checkpoint
#### Path Parameters
- **id** (string) - Required - The unique identifier of the pipeline.
### Response
#### Success Response (200)
- **checkpoint_data** (object) - The data associated with the current checkpoint.
```
--------------------------------
### Usage Record Addition
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Demonstrates how Usage objects support addition for aggregating across multi-step operations by summing integer fields and handling optional fields.
```protobuf
usage_a + usage_b -> Usage
-- Sums integer fields.
-- For optional fields: if either side is non-None, sum them (treating None as 0).
-- If both sides are None for an optional field, the result is None.
```
--------------------------------
### Get Context
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Retrieves the current context, which is a key-value store representing the state of the pipeline execution. This can be useful for debugging and understanding runtime variables.
```APIDOC
## GET /pipelines/{id}/context
### Description
Get current context key-value store.
### Method
GET
### Endpoint
/pipelines/{id}/context
#### Path Parameters
- **id** (string) - Required - The unique identifier of the pipeline.
### Response
#### Success Response (200)
- **context** (object) - A key-value store representing the pipeline's current context.
```
--------------------------------
### Implement Attractor with a Coding Agent
Source: https://github.com/strongdm/attractor/blob/main/README.md
Use this prompt with a modern coding agent to implement Attractor based on the provided specification.
```bash
codeagent> Implement Attractor as described by https://github.com/strongdm/attractor
```
--------------------------------
### Get Pending Questions
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Retrieves any pending questions that require human interaction for a specific pipeline. This is used for pipelines that involve human gates.
```APIDOC
## GET /pipelines/{id}/questions
### Description
Get pending human interaction questions.
### Method
GET
### Endpoint
/pipelines/{id}/questions
#### Path Parameters
- **id** (string) - Required - The unique identifier of the pipeline.
### Response
#### Success Response (200)
- **questions** (array) - A list of pending questions, each with a unique question ID (qid).
```
--------------------------------
### Get Rendered Graph
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Retrieves a visual representation (SVG) of the pipeline's graph. This is useful for debugging and understanding the pipeline's structure.
```APIDOC
## GET /pipelines/{id}/graph
### Description
Get rendered graph visualization (SVG).
### Method
GET
### Endpoint
/pipelines/{id}/graph
#### Path Parameters
- **id** (string) - Required - The unique identifier of the pipeline.
### Response
#### Success Response (200)
- **SVG content** - The SVG markup for the pipeline graph.
```
--------------------------------
### Tool Execution Pipeline Stages
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Outlines the sequential steps involved in executing a tool, from lookup and validation to execution, truncation, and emission.
```plaintext
1. LOOKUP -- find the RegisteredTool by name
2. VALIDATE -- parse and validate arguments against JSON Schema
3. EXECUTE -- call executor with (arguments, execution_env)
4. TRUNCATE -- apply output size limits (Section 5)
5. EMIT -- emit TOOL_CALL_END event with full output
6. RETURN -- return truncated output as ToolResult
```
--------------------------------
### ConsoleInterviewer (CLI) Implementation
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
This interviewer interacts with the user via the command line, displaying formatted prompts and reading input. It supports timeouts through non-blocking reads.
```strongdm-attractor
ConsoleInterviewer:
FUNCTION ask(question) -> Answer:
print("[?] " + question.text)
IF question.type == MULTIPLE_CHOICE:
FOR EACH option IN question.options:
print(" [" + option.key + "] " + option.label)
response = read_input("Select: ")
RETURN find_matching_option(response, question.options)
IF question.type == YES_NO:
response = read_input("[Y/N]: ")
RETURN Answer(value=YES if response is "y" ELSE NO)
IF question.type == FREEFORM:
response = read_input("> ")
RETURN Answer(text=response)
```
--------------------------------
### Pipeline Preparation Function
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Illustrates the process of preparing a pipeline by parsing DOT source, applying a series of transforms, and then validating the resulting graph.
```pseudocode
FUNCTION prepare_pipeline(dot_source):
graph = parse(dot_source)
FOR EACH transform IN transforms:
graph = transform.apply(graph)
diagnostics = validate(graph)
RETURN (graph, diagnostics)
```
--------------------------------
### Registering a Custom Tool with an OpenAI Profile
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Demonstrates how to add a custom tool, 'run_tests', to an existing OpenAI profile. Custom tools override profile tools if names collide.
```python
profile = create_openai_profile(model = "gpt-5.2-codex")
-- Add a custom tool on top of the profile
profile.tool_registry.register(RegisteredTool(
definition = ToolDefinition(
name = "run_tests",
description = "Run the project's test suite",
parameters = { "type": "object", "properties": { "filter": { "type": "string" } } }
),
executor = run_tests_function
))
```
--------------------------------
### Submit Pipeline
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Submits a DOT source to the Attractor engine for execution. This is the primary endpoint for starting a new pipeline. It returns a unique identifier for the submitted pipeline.
```APIDOC
## POST /pipelines
### Description
Submit a DOT source and start execution. Returns pipeline ID.
### Method
POST
### Endpoint
/pipelines
### Request Body
- **dot_source** (string) - Required - The DOT source code for the pipeline.
### Response
#### Success Response (200)
- **pipeline_id** (string) - The unique identifier for the submitted pipeline.
### Request Example
{
"dot_source": "digraph G { start -> end }"
}
```
--------------------------------
### Basic Generation Across Providers
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Tests basic text generation by iterating through supported providers (Anthropic, OpenAI, Gemini). Asserts that the generated text is not empty and usage statistics are positive, and that the finish reason is 'stop'.
```pseudocode
-- 1. Basic generation across all providers
FOR EACH provider IN ["anthropic", "openai", "gemini"]:
result = generate(
model = get_latest_model(provider).id,
prompt = "Say hello in one sentence.",
max_tokens = 100,
provider = provider
)
ASSERT result.text is not empty
ASSERT result.usage.input_tokens > 0
ASSERT result.usage.output_tokens > 0
ASSERT result.finish_reason.reason == "stop"
```
--------------------------------
### Generation with Tools
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Generates a response using a specified model, system prompt, and tools, with a limit on tool rounds. Prints the final text, number of steps, and total token usage.
```python
result = generate(
model = "claude-opus-4-6",
system = "You are a helpful assistant with access to weather data.",
prompt = "What is the weather in San Francisco?",
tools = [weather_tool],
max_tool_rounds = 5
)
PRINT(result.text) -- final text after all tool rounds
PRINT(LENGTH(result.steps)) -- number of steps taken
PRINT(result.total_usage.total_tokens) -- aggregated token count
```
--------------------------------
### Custom Handler Definition and Registration
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
This snippet demonstrates how to define a custom handler by implementing the Handler interface and then registering it with the registry. It also shows how to reference the custom handler in a DOT file.
```DOT
-- Define a custom handler
MyCustomHandler:
FUNCTION execute(node, context, graph, logs_root) -> Outcome:
-- Custom logic here
RETURN Outcome(status=SUCCESS)
-- Register it
registry.register("my_custom_type", MyCustomHandler())
-- Reference in DOT file
my_node [type="my_custom_type", shape=box, custom_attr="value"]
```
--------------------------------
### CallbackInterviewer Implementation
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
This interviewer delegates question answering to a provided callback function. It is useful for integrating with external systems like Slack, web UIs, or APIs.
```strongdm-attractor
CallbackInterviewer:
callback : Function(Question) -> Answer
FUNCTION ask(question) -> Answer:
RETURN callback(question)
```
--------------------------------
### Context Structure and Operations
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Defines the thread-safe Context structure, including its key-value store, lock, and log. It outlines functions for setting, getting, logging, snapshotting, cloning, and applying updates to the context.
```Pseudocode
Context:
values : Map -- key-value store
lock : ReadWriteLock -- thread safety for parallel access
logs : List -- append-only run log
FUNCTION set(key, value):
ACQUIRE write lock
values[key] = value
RELEASE write lock
FUNCTION get(key, default=NONE) -> Any:
ACQUIRE read lock
result = values.get(key, default)
RELEASE read lock
RETURN result
FUNCTION get_string(key, default="") -> String:
value = get(key)
IF value is NONE: RETURN default
RETURN string(value)
FUNCTION append_log(entry):
ACQUIRE write lock
logs.append(entry)
RELEASE write lock
FUNCTION snapshot() -> Map:
-- Returns a serializable deep copy of all values
ACQUIRE read lock
result = deep_copy(values)
RELEASE read lock
RETURN result
FUNCTION clone() -> Context:
-- Deep copy for parallel branch isolation
ACQUIRE read lock
new_context = new Context()
new_context.values = deep_copy(values)
new_context.logs = copy(logs)
RELEASE read lock
RETURN new_context
FUNCTION apply_updates(updates):
-- Merge a dictionary of updates into the context
ACQUIRE write lock
FOR EACH (key, value) IN updates:
values[key] = value
RELEASE write lock
```
--------------------------------
### stream() Function Usage Example
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
The `stream()` function is the primary streaming generation function, yielding events incrementally. It accepts the same parameters as `generate()` and provides a `StreamResult` object for iterating over events and accessing the final response.
```Python
result = stream(
model = "claude-opus-4-6",
prompt = "Write a haiku about coding"
)
FOR EACH event IN result:
IF event.type == TEXT_DELTA:
PRINT(event.delta)
-- After iteration, the full response is available:
response = result.response()
```
--------------------------------
### RecordingInterviewer Implementation
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
This interviewer wraps another interviewer and records all question-answer pairs. It is used for replay, debugging, and creating audit trails.
```strongdm-attractor
RecordingInterviewer:
inner : Interviewer
recordings : List<(Question, Answer)>
FUNCTION ask(question) -> Answer:
answer = inner.ask(question)
recordings.append((question, answer))
RETURN answer
```
--------------------------------
### Configure OpenAI-Compatible Adapter
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Use this adapter for third-party services that expose an OpenAI-compatible Chat Completions API. Ensure you provide the correct API key and base URL.
```python
adapter = OpenAICompatibleAdapter(
api_key = "...",
base_url = "https://my-vllm-instance.example.com/v1"
)
```
--------------------------------
### Optional ProviderAdapter Methods
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Provides optional methods for provider adapters, such as resource cleanup, initialization validation, and tool choice support.
```pseudocode
FUNCTION close() -> Void
-- Release resources (HTTP connections, etc.). Called by Client.close().
FUNCTION initialize() -> Void
-- Validate configuration on startup. Called by Client on registration.
FUNCTION supports_tool_choice(mode: String) -> Boolean
-- Query whether a particular tool choice mode is supported.
```
--------------------------------
### Implement WaitForHumanHandler
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Handles human interaction by presenting choices derived from outgoing edges and waiting for user selection. Includes logic for timeouts and skips.
```Go
WaitForHumanHandler:
interviewer : Interviewer -- the human interaction frontend
FUNCTION execute(node, context, graph, logs_root) -> Outcome:
-- 1. Derive choices from outgoing edges
edges = graph.outgoing_edges(node.id)
choices = []
FOR EACH edge IN edges:
label = edge.label OR edge.to_node
key = parse_accelerator_key(label)
choices.append(Choice(key=key, label=label, to=edge.to_node))
IF choices is empty:
RETURN Outcome(status=FAIL, failure_reason="No outgoing edges for human gate")
-- 2. Build question from choices
options = [Option(key=c.key, label=c.label) FOR c IN choices]
question = Question(
text=node.label OR "Select an option:",
type=MULTIPLE_CHOICE,
options=options,
stage=node.id
)
-- 3. Present to interviewer and wait for answer
answer = interviewer.ask(question)
-- 4. Handle timeout/skip
IF answer is TIMEOUT:
default_choice = node.attrs["human.default_choice"]
IF default_choice exists:
-- Use default
ELSE:
RETURN Outcome(status=RETRY, failure_reason="human gate timeout, no default")
IF answer is SKIPPED:
RETURN Outcome(status=FAIL, failure_reason="human skipped interaction")
-- 5. Find matching choice
selected = find_choice_matching(answer, choices)
IF selected is NONE:
selected = choices[0] -- fallback to first
-- 6. Record in context and return
RETURN Outcome(
status=SUCCESS,
suggested_next_ids=[selected.to],
context_updates={
"human.gate.selected": selected.key,
"human.gate.label": selected.label
}
)
```
--------------------------------
### Gemini Tool Definition Translation Table
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Illustrates the mapping of tool definition fields from a generic SDK format to Gemini's specific structure.
```markdown
| SDK Format | OpenAI | Anthropic | Gemini |
|-------------------------|----------------------------------------------------|-------------------------------------------------|-----------------------------------------------------|
| Tool.name | tools[].function.name | tools[].name | tools[].functionDeclarations[].name |
| Tool.description | tools[].function.description | tools[].description | tools[].functionDeclarations[].description |
| Tool.parameters | tools[].function.parameters | tools[].input_schema | tools[].functionDeclarations[].parameters |
| Wrapper structure | `{"type":"function","function":{...}}` | `{"name":...,"description":...,"input_schema":...}` | `{"functionDeclarations":[{...}]}` |
```
--------------------------------
### Handler Registry Implementation
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Details the HandlerRegistry structure, including its handler map and default handler, and the functions for registering and resolving handlers.
```Go
HandlerRegistry:
handlers : Map -- type string -> handler instance
default_handler : Handler -- fallback handler (typically codergen)
FUNCTION register(type_string, handler):
handlers[type_string] = handler
-- Registering for an already-registered type replaces the previous handler
FUNCTION resolve(node) -> Handler:
-- 1. Explicit type attribute
IF node.type is not empty AND node.type IN handlers:
RETURN handlers[node.type]
-- 2. Shape-based resolution
handler_type = SHAPE_TO_TYPE[node.shape]
IF handler_type IN handlers:
RETURN handlers[handler_type]
-- 3. Default
RETURN default_handler
```
--------------------------------
### Execute Tool and Handle Results
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
This snippet demonstrates the core logic for executing a tool, truncating its output, and emitting events. It includes error handling for invalid arguments and execution errors.
```python
error_msg = "Invalid arguments for tool: " + tool_call.name
session.emit(TOOL_CALL_END, call_id = tool_call.id, error = error_msg)
RETURN ToolResult(tool_call_id = tool_call.id, content = error_msg, is_error = true)
-- Execute via execution environment
TRY:
raw_output = registered.executor(tool_call.arguments, session.execution_env)
-- Truncate output before sending to LLM (character-based first, then line-based)
truncated_output = truncate_tool_output(raw_output, tool_call.name, session.config)
-- Emit full output via event stream (not truncated)
session.emit(TOOL_CALL_END, call_id = tool_call.id, output = raw_output)
RETURN ToolResult(
tool_call_id = tool_call.id,
content = truncated_output,
is_error = false
)
CATCH error:
error_msg = "Tool error (" + tool_call.name + "): " + str(error)
session.emit(TOOL_CALL_END, call_id = tool_call.id, error = error_msg)
RETURN ToolResult(tool_call_id = tool_call.id, content = error_msg, is_error = true)
```
--------------------------------
### QueueInterviewer Implementation
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
This interviewer reads answers from a pre-filled queue. It is used for deterministic testing and replaying interactions.
```strongdm-attractor
QueueInterviewer:
answers : Queue
FUNCTION ask(question) -> Answer:
IF answers is not empty:
RETURN answers.dequeue()
RETURN Answer(value=SKIPPED)
```
--------------------------------
### Retry-After Header Handling
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Describes how the library respects the `Retry-After` header from providers, especially for 429 responses. It prioritizes the provider's suggested delay but avoids excessive waits.
```text
- If `Retry-After` is less than `max_delay`, use the provider's delay instead of the calculated backoff.
- If `Retry-After` exceeds `max_delay`, do NOT retry. Raise the error immediately with `retry_after` set on the exception. This prevents silently waiting minutes for a rate limit to clear.
```
--------------------------------
### Apply Patch v4a: Add File Operation
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
This snippet demonstrates how to add a new file using the v4a patch format. All lines to be added to the new file must be prefixed with a '+'.
```text
*** Begin Patch
*** Add File: src/utils/helpers.py
+def greet(name):
+ return f"Hello, {name}!"
*** End Patch
```
--------------------------------
### Injecting Messages into Agent Conversation
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
Use `session.steer` to queue a message for injection after the current tool round. If the agent is idle, the message is delivered on the next submit(). Use `session.follow_up` to queue a message to be processed after the current input is fully handled, triggering a new processing cycle.
```swift
session.steer(message: String)
-- Queue a message to be injected after the current tool round completes.
-- The message becomes a SteeringTurn in the history, converted to a
-- user message for the LLM on the next call.
-- If the agent is idle, the message is delivered on the next submit().
session.follow_up(message: String)
-- Queue a message to be processed after the current input is fully handled
-- (model has produced a text-only response). Triggers a new processing cycle.
```
--------------------------------
### Registering a Custom Transform
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Shows how to register a custom transform implementation with the Attractor runner.
```python
runner.register_transform(MyCustomTransform())
```
--------------------------------
### Checkpoint Data Structure and Functions
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Defines the structure of a checkpoint, including timestamp, node information, and context values. Includes functions for saving the checkpoint to a file and loading it.
```Go
Checkpoint:
timestamp : Timestamp -- when this checkpoint was created
current_node : String -- ID of the last completed node
completed_nodes : List -- IDs of all completed nodes in order
node_retries : Map -- retry counters per node
context_values : Map -- serialized snapshot of the context
logs : List -- run log entries
FUNCTION save(path):
-- Serialize to JSON and write to filesystem
data = {
"timestamp": timestamp,
"current_node": current_node,
"completed_nodes": completed_nodes,
"node_retries": node_retries,
"context": serialize_to_json(context_values),
"logs": logs
}
write_json_file(path, data)
FUNCTION load(path) -> Checkpoint:
-- Deserialize from JSON file
data = read_json_file(path)
RETURN new Checkpoint from data
```
--------------------------------
### Event Consumption Patterns
Source: https://github.com/strongdm/attractor/blob/main/attractor-spec.md
Demonstrates two common patterns for consuming events emitted by the attractor engine: the observer pattern using callbacks and the stream pattern for asynchronous runtimes.
```pseudocode
-- Observer pattern
runner.on_event = FUNCTION(event):
log(event.description)
-- Stream pattern (for async runtimes)
FOR EACH event IN pipeline.events():
process(event)
```
--------------------------------
### Applying Retry Logic with a Standalone Utility
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Demonstrates how to manually apply retry behavior to low-level API calls using a standalone `retry()` utility. This is an alternative to relying on high-level functions with built-in retries.
```python
response = retry(
FUNCTION: client.complete(request),
policy = RetryPolicy(max_retries = 3)
)
```
--------------------------------
### Apply Patch v4a: Update and Rename File Operation
Source: https://github.com/strongdm/attractor/blob/main/coding-agent-loop-spec.md
This snippet demonstrates how to update and rename a file simultaneously using the v4a patch format. The new path is specified with 'Move to:'.
```text
*** Begin Patch
*** Update File: old_name.py
*** Move to: new_name.py
@@ import os
import sys
-import old_dep
+import new_dep
*** End Patch
```
--------------------------------
### Tool Handler with Context Injection
Source: https://github.com/strongdm/attractor/blob/main/unified-llm-spec.md
Demonstrates a tool handler that accepts injected context arguments such as the conversation history, a cancellation signal, and the tool call ID.
```python
FUNCTION my_tool(
query : String, -- tool parameter
messages : List, -- injected: current conversation
abort_signal : AbortSignal, -- injected: cancellation signal
tool_call_id : String -- injected: ID of this call
) -> String:
...
```