Python quickstart

Before you start

If you only need the smallest runnable example, start with First workflow in five minutes. This page adds install verification, trace reading, optional OpenAI, error handling, and remote server mode in one sitting.

Read 03 Anatomy of a workflow so terms like step, run input, and result.output are already familiar.

Concept

The Python SDK exposes idiomatic Agent and Workflow types that serialize to the ArcFlow workflow specification and call the in-process Rust runtime by default. You declare agents and step order in Python; arcflow-core executes the graph, invokes providers, and emits trace events.

Without a provider= argument, run() uses the default in-process agent backend. No OPENAI_API_KEY is required on that path. Passing runtime="http://localhost:8080" targets arcflow-server instead while keeping the same declaration code.

Install

From the repository root:

cd sdk-python
pip install maturin
maturin develop
pip install -e ".[dev]"

Confirm the package loads:

python -c "from arcflow import Agent, Workflow; print('import ok')"

See Install and build for platform notes (macOS, Linux, Windows) and troubleshooting.

Example: agent and workflow

Save as quickstart.py:

from arcflow import Agent, Workflow

researcher = Agent(
 name="researcher",
 role="research",
 instructions="Research the given topic and list key facts.",
)
writer = Agent(
 name="writer",
 role="write",
 instructions="Turn the research into a short paragraph.",
)

workflow = Workflow("research_pipeline")
workflow.step(researcher)
workflow.step(writer)

result = workflow.run("Analyze renewable energy trends")

print(result.output)
print(result.run_id)
print(result.step_count)
print(result.status)

Run:

python quickstart.py

You should see step_count == 2, status of completed, and non-empty output.

Reading the result

WorkflowResult fields used most often:

for event in result.trace_events:
 print(event.get("event_kind"), event.get("sequence"))

Trace payloads follow trace data policy rules: no raw prompts, tool values, or credentials. See Trace data policy rules and Trace events (normative) for event shapes.

Structured trace via workflow.trace()

After run(), the workflow object exposes a parsed trace:

trace = workflow.trace()
print(trace.summary())
print(trace.status)
print(len(trace.steps))
print(trace.total_tokens_consumed)

Calling trace() before run() raises TraceNotFoundError.

Optional: real LLM with OpenAI

When you have an API key, pass a provider to run():

import os

from arcflow import Agent, OpenAI, Workflow

os.environ.setdefault("OPENAI_API_KEY", "sk-your-key-here")

wf = Workflow("demo")
wf.step(Agent(name="writer", role="author", instructions="Summarize in three sentences."))

result = wf.run(
 "Quantum networking",
 provider=OpenAI(model="gpt-4o"),
)
print(result.output)

Supported environment variables for other providers: ANTHROPIC_API_KEY, GEMINI_API_KEY with Anthropic and Gemini classes from arcflow.

Provider failures raise ProviderExecutionError with a provider_id when set. Configuration mistakes raise ProviderConfigurationError before the run starts.

See Provider configuration for model params and retry behavior.

Remote server mode (optional)

To target arcflow-server instead of the in-process runtime:

import os

os.environ["ARCFLOW_SERVER_API_KEY"] = "dev-secret"

from arcflow import Agent, Workflow

wf = Workflow("demo", runtime="http://localhost:8080")
wf.step(Agent(name="writer", role="author", instructions="Summarize."))
result = wf.run("hello")
print(result.output)

See Server API quickstart and Integrating track.

Common errors

Messages use the format [ArcFlow] <what happened>. <what to do>.

Verify

Next