@arythmatic/flow-client-py

Official Python client for Arythmatic Flow — the API platform for testing, workflows, AI, monitoring, and security scanning.

Install

pip install flow-client

Quick Start

from flow_client import FlowClient

flow = FlowClient()

# Authenticate
flow.login({"email": "you@example.com", "password": "secret"})

# Set active team
flow.set_team_id("your-team-id")

# Run a stored collection
collections = flow.collections.list()
result = flow.collections.run(collections[0]["id"])
print(result)

Features

• HTTP Client — requests-like interface with interceptors, timeouts, and auto-auth headers.
• Collections — list, run, and inspect stored request collections.
• Workflows — execute DAG-based workflows with custom inputs.
• AI — generate requests from descriptions, suggest assertions, manage conversations.
• Monitoring — create uptime/performance monitors and trigger on-demand checks.
• Security Scanning — run OWASP-style vulnerability scans on requests, collections, or URLs.
• Proxy Execution — send arbitrary HTTP requests through Flow's engine (CORS-free, with variables, scripts, and assertions).

Configuration

flow = FlowClient(
    base_url="https://flow.arythmatic.cloud",  # or your self-hosted instance
    access_token="...",   # optional: pre-authenticated
    refresh_token="...",  # optional
    team_id="...",        # optional: default team
    timeout=30000,        # request timeout in ms
    headers={},           # extra default headers
)

Authentication

# Login (stores tokens automatically)
user_and_teams = flow.login({"email": email, "password": password})

# Register
user_and_team = flow.register({"email": email, "password": password, "name": name})

# Refresh
flow.refresh()

# Get current user
me = flow.auth.me()

Collections

collections = flow.collections.list()

# Run the entire collection
run = flow.collections.run(collection_id, {
    "mode": "sequential",       # or "parallel"
    "environmentId": "env-id",
    "stopOnFailure": True,
})

# Run only specific requests from the collection
run_subset = flow.collections.run(collection_id, {
    "requestIds": ["req-id-1", "req-id-2"],
})

# Run a single request from a collection
run_one = flow.collections.run_one(collection_id, "req-id-1")

runs = flow.collections.runs(collection_id)

Workflows

workflows = flow.workflows.list()
result = flow.workflows.run(workflow_id, {
    "environmentId": "env-id",
    "input": {"userId": "123"},
})
history = flow.workflows.runs(workflow_id)

# Get a specific run or resume a failed one
run = flow.workflows.run_detail(run_id)
resumed = flow.workflows.resume_run(run_id, from_node_id="node-1")

AI

# Generate request definitions from text
result = flow.ai.generate({
    "input": "A todo list API with CRUD",
    "type": "description",
})

# Suggest assertions
result = flow.ai.suggest_assertions({
    "status": 200,
    "headers": {},
    "body": "...",
    "url": "https://api.example.com/users",
    "method": "GET",
})

# Token usage
usage = flow.ai.usage()

Monitors

monitor = flow.monitors.create({
    "name": "API Health",
    "type": "standalone",
    "url": "https://api.example.com/health",
    "method": "GET",
    "expectedStatus": 200,
    "interval": "30m",
})
flow.monitors.check(monitor["id"])

Security Scans

# Scan a stored request
flow.scans.scan_request(request_id)

# Scan a collection
flow.scans.scan_collection(collection_id)

# Scan any URL
flow.scans.scan_url("https://api.example.com/users")

# Get results
report = flow.scans.get(scan_id)

Environments

envs = flow.environments.list()
env = flow.environments.create("Staging", color="#3b82f6")
flow.environments.set_vars(env["id"], [{"k": "BASE_URL", "v": "https://staging.example.com"}])

Data Tables

tables = flow.data_tables.list()
table = flow.data_tables.create("Users", columns=["id", "name", "role"])
flow.data_tables.update(table["id"], rows=[[1, "Alice", "admin"]])

Request Templates

tpls = flow.request_templates.list()
tpl = flow.request_templates.create("Get User", method="GET", path="/users/{{id}}")

Comments

comments = flow.comments.list_for_request(request_id)
flow.comments.create(request_id, "Looks good!")

Audit Log

log = flow.audit_log.list(entity="collection", limit=50)
users = flow.audit_log.users()

Performance Tests

result = flow.perf_tests.run("https://api.example.com/health", concurrency=20, total_requests=200)
runs = flow.perf_tests.list()

Team Management

my_teams = flow.teams.list()
members = flow.teams.members(team_id)
flow.teams.invite(team_id, "new@example.com", role="admin")
flow.teams.accept_invitation(token)

History

entries = flow.history.list(collection_id=collection_id, limit=20)
entry = flow.history.get(entry_id)
flow.history.delete_many(older_than_days=30)

Import / Export

result = flow.import_export.import_collection(postman_json)
flow.import_export.export_collection(collection_id, format="openapi")
flow.import_export.export_cicd(collection_id, platform="github", schedule="0 9 * * *")

Workflow Folders & Templates

folders = flow.workflow_folders.list()
flow.workflow_folders.create("Onboarding", parent_id=parent_folder_id)

templates = flow.workflow_templates.list()
flow.workflow_templates.use(template_id, name="My Onboarding Flow")

MCP Servers

templates = flow.mcp_servers.templates()
servers = flow.mcp_servers.list()
flow.mcp_servers.create("slack", type="template", credentials={"token": "xoxb-..."})
flow.mcp_servers.reconnect(server_id)

Proxy / Arbitrary Requests

Send any HTTP request through Flow's execution engine:

res = flow.requests.execute({
    "method": "POST",
    "url": "https://api.example.com/users",
    "headers": {"Content-Type": "application/json"},
    "body": '{"name": "Alice"}',
    "envVars": [{"key": "BASE_URL", "value": "https://api.example.com"}],
    "assertions": [
        {"type": "status", "operator": "eq", "expected": 201},
    ],
})

Shorthand helpers are also available:

flow.requests.get(url, opts)
flow.requests.post(url, opts)
# put, patch, delete

Interceptors

flow.interceptors.request.use(lambda cfg: {
    **cfg,
    "headers": {**cfg["headers"], "X-Custom": "value"}
})

flow.interceptors.response.use(
    lambda res: res,
    lambda err: print(f"Request failed: {err}") or (_ for _ in ()).throw(err)
)

Error Handling

from flow_client import FlowError, FlowAuthError

try:
    flow.collections.list()
except FlowAuthError:
    flow.refresh()
except FlowError as err:
    print(err.status, err.code, err.response)

CLI

A small CLI is included for quick tasks:

# Login
python -m flow_client login -u http://localhost:3001 -e you@example.com -p secret

# List collections
python -m flow_client collections list

# Run a collection
python -m flow_client collections run <collection-id> -e <env-id>

# View runs
python -m flow_client runs list <collection-id>
python -m flow_client runs get <collection-id> <run-id>

License

MIT