Documentation Index
Fetch the complete documentation index at: https://mintlify.com/8BitTacoSupreme/flowstate/llms.txt
Use this file to discover all available pages before exploring further.
ClaudeBridge is the integration layer between FlowState’s Python pipeline and the Claude Code CLI. It wraps the claude --print command — Claude Code’s non-interactive mode — and exposes a typed Python API for running prompts with system personas, scoped tool permissions, model selection, and spend caps. Every LLM-backed pipeline step routes through a single ClaudeBridge instance created by the orchestrator.
BridgeConfig
BridgeConfig is a dataclass that holds all per-process settings. The orchestrator creates one config for the entire pipeline run; individual adapters can override specific fields per call.
# flowstate/bridge.py
@dataclass
class BridgeConfig:
claude_bin: str | None = None # path to claude binary; auto-detected if None
project_root: Path = field(default_factory=Path.cwd) # subprocess working directory
timeout: int = 300 # seconds before subprocess.TimeoutExpired
allowed_tools: list[str] = field(default_factory=list) # default tool permissions
max_turns: int = 10 # default agentic turn limit
model: str | None = None # model override (None = claude default)
max_budget_usd: float | None = None # spend cap per call (None = no cap)
effort: str | None = None # effort level hint
BridgeConfig.__post_init__ calls _find_claude() automatically when claude_bin is None, so you rarely need to set it explicitly.
Binary Auto-Detection
FlowState locates the claude binary using the following priority order:
FLOWSTATE_CLAUDE_BIN environment variable
If set and the path exists, this is used unconditionally — useful in containers or CI where claude is not on PATH.export FLOWSTATE_CLAUDE_BIN=/opt/claude-code/bin/claude
PATH lookup via shutil.which
Standard PATH resolution. Works for most local installations.
Common install locations
Falls back to three hardcoded candidates in order:# flowstate/bridge.py
candidates = [
Path.home() / ".local" / "bin" / "claude", # npm global (Linux/macOS)
Path("/usr/local/bin/claude"), # system-wide
Path("/opt/homebrew/bin/claude"), # Homebrew (macOS ARM)
]
If none of these resolve to an existing file, claude_bin is set to an empty string. The available property reflects this:
# flowstate/bridge.py
@property
def available(self) -> bool:
return bool(self.config.claude_bin)
The orchestrator checks bridge.available after creating the instance. If it returns False in a live (non-dry-run) run, the pipeline automatically falls back to dry-run mode:
# flowstate/orchestrator.py
if not dry_run and not bridge.available:
bridge = ClaudeBridge(config=bridge.config, dry_run=True)
ClaudeBridge.run()
run() is the primary method. It constructs the claude CLI invocation and captures stdout/stderr.
# flowstate/bridge.py
def run(
self,
prompt: str,
*,
system_prompt: str | None = None,
allowed_tools: list[str] | None = None,
output_format: str = "text",
max_turns: int | None = None,
model: str | None = _SENTINEL,
) -> BridgeResult:
Parameters:
| Parameter | Type | Description |
|---|
prompt | str | The prompt text — passed as a positional argument to claude, not --prompt |
system_prompt | str | None | Injected via --system-prompt; used for researcher/advisor personas |
allowed_tools | list[str] | None | Per-call tool overrides; falls back to config.allowed_tools if None |
output_format | str | "text" (default) or "json" — controls --output-format flag |
max_turns | int | None | Per-call turn limit; falls back to config.max_turns if None |
model | str | None | Per-call model override. Defaults to _SENTINEL (use config.model). Pass None explicitly to suppress --model for this call. |
How the CLI is invoked
The method builds the command list incrementally and appends -- before the prompt to prevent any flags embedded in the prompt from being interpreted as CLI flags:
# flowstate/bridge.py
cmd = [self.config.claude_bin, "--print"]
if output_format == "json":
cmd.extend(["--output-format", "json"])
turns = max_turns or self.config.max_turns
cmd.extend(["--max-turns", str(turns)])
tools = allowed_tools or self.config.allowed_tools
if tools:
cmd.extend(["--allowedTools", ",".join(tools)])
if system_prompt:
cmd.extend(["--system-prompt", system_prompt])
effective_model = model if model is not _SENTINEL else self.config.model
if effective_model:
cmd.extend(["--model", effective_model])
if self.config.max_budget_usd is not None:
cmd.extend(["--max-budget-usd", str(self.config.max_budget_usd)])
if self.config.effort:
cmd.extend(["--effort", self.config.effort])
# "--" separates CLI flags from the positional prompt
cmd.append("--")
cmd.append(prompt)
This produces invocations like:
claude --print \
--max-turns 10 \
--allowedTools "Read,WebSearch" \
--system-prompt "You are a senior researcher..." \
--model claude-opus-4-5 \
--max-budget-usd 0.50 \
-- "Research topic: Kafka Streams stateful processing..."
The CLAUDECODE env var
Before the subprocess is launched, CLAUDECODE is removed from the environment:
# flowstate/bridge.py
env = {**os.environ}
env.pop("CLAUDECODE", None)
The CLAUDECODE variable is set by Claude Code when it starts a session. Keeping it set would prevent nested claude invocations from running correctly — removing it allows FlowState’s bridge calls to work even when flowstate init is run from inside a Claude Code session.
BridgeResult
Every run() call returns a BridgeResult:
# flowstate/bridge.py
@dataclass
class BridgeResult:
success: bool # True if exit_code == 0
output: str # captured stdout
exit_code: int = 0 # subprocess return code
error: str | None = None # stderr content (only when exit_code != 0)
The orchestrator’s _run_step() function uses result.success to decide whether to mark the tool COMPLETED or BLOCKED.
Error conditions
run() handles two exceptional cases and returns a failed BridgeResult for each rather than raising:
subprocess.TimeoutExpired — returns exit_code=-1 and error="claude CLI timed out after Ns"
FileNotFoundError — returns exit_code=-1 and error="claude CLI not found at: <path>"
ClaudeBridge.invoke_skill()
invoke_skill() wraps run() for Claude Code slash-command skills (e.g., gsd:new-project). It formats the skill as a /skill args prompt and grants a broad set of filesystem tools:
# flowstate/bridge.py
def invoke_skill(self, skill: str, args: str = "") -> BridgeResult:
"""Invoke a Claude Code skill (e.g., 'gsd:new-project')."""
prompt = f"/{skill}"
if args:
prompt += f" {args}"
return self.run(
prompt,
allowed_tools=["Read", "Write", "Edit", "Bash", "Glob", "Grep"],
max_turns=15,
)
Dry-Run Mode
When dry_run=True, run() returns immediately with a mock result and never spawns a subprocess:
# flowstate/bridge.py
if self.dry_run:
return BridgeResult(
success=True,
output=f"[dry-run] claude prompt ({len(prompt)} chars): {prompt[:120]}...",
)
The mock output includes the prompt length and a 120-character preview so tests can assert the prompt was constructed correctly without needing the real CLI.
Full Usage Example
from pathlib import Path
from flowstate.bridge import BridgeConfig, ClaudeBridge
# Create a config for this project
config = BridgeConfig(
project_root=Path("/my/project"),
timeout=300,
max_turns=10,
model="claude-opus-4-5",
max_budget_usd=1.00,
)
bridge = ClaudeBridge(config=config)
# Check if claude CLI was found
if not bridge.available:
print("claude CLI not found — install Claude Code or set FLOWSTATE_CLAUDE_BIN")
raise SystemExit(1)
# Run a research prompt with a custom system prompt and tool permissions
result = bridge.run(
"Research the trade-offs between Kafka Streams and Apache Flink for stateful processing.",
system_prompt="You are a senior distributed systems researcher. Be concise and cite specifics.",
allowed_tools=["WebSearch", "Read"],
max_turns=5,
)
if result.success:
print(result.output)
else:
print(f"Bridge call failed (exit {result.exit_code}): {result.error}")
# Invoke a GSD skill
skill_result = bridge.invoke_skill("gsd:new-project", args="--auto")
Use flowstate check to verify that ClaudeBridge can locate the claude binary and to see the effective timeout and turn settings before running a full pipeline:flowstate check
# claude CLI found: /Users/you/.local/bin/claude
# Timeout: 300s | Max turns: 10
Creating the Bridge in the Orchestrator
The orchestrator builds a ClaudeBridge using project preferences stored in flowstate.json, forwarding any user-specified model, budget, or effort overrides:
# flowstate/orchestrator.py
def _make_bridge(root: Path, dry_run: bool, preferences=None) -> ClaudeBridge:
kwargs = {"project_root": root}
if preferences:
if preferences.model:
kwargs["model"] = preferences.model
if preferences.max_budget_usd is not None:
kwargs["max_budget_usd"] = preferences.max_budget_usd
if preferences.effort:
kwargs["effort"] = preferences.effort
config = BridgeConfig(**kwargs)
return ClaudeBridge(config=config, dry_run=dry_run)
This single bridge instance is shared across the ResearchAdapter, StrategyAdapter, and GSDAdapter for the entire pipeline run, so all LLM calls in one pipeline share the same model and budget settings.