The easiest way. Creates everything internally.

from splinter import Splinter

s = Splinter(openai_key="sk-...", max_budget=10.0)
result = await s.run("researcher", "Find AI trends")
print(f"Cost: ${s.cost:.4f}")

Run multiple agents with dependencies and shared limits.

from splinter.workflow import Workflow
from splinter.types import AgentConfig, ExecutionLimits, LLMProvider

workflow = Workflow(
    workflow_id="pipeline",
    limits=ExecutionLimits(max_budget=20.0),
    checkpoint_enabled=True,
)

workflow.add_agent(AgentConfig(
    agent_id="researcher",
    provider=LLMProvider.OPENAI,
    model="gpt-4o",
    system_prompt="Research topics. Output JSON.",
))

workflow.add_agent(AgentConfig(
    agent_id="writer",
    provider=LLMProvider.OPENAI,
    model="gpt-4o",
    system_prompt="Write articles. Output JSON.",
))

workflow.add_step("researcher")
workflow.add_step("writer", depends_on=["researcher"])

result = await workflow.run(initial_state={"topic": "AI"})

Build agents with the fluent API.

from splinter.workflow import AgentBuilder
from splinter.types import LLMProvider

agent = (
    AgentBuilder("researcher")
    .with_provider(LLMProvider.OPENAI, "gpt-4o")
    .with_system_prompt("Research topics. Output JSON.")
    .with_tools(["web_search"])
    .with_state_ownership(["research.*"])
    .build(gateway)
)

result = await agent.run(task="Research AI trends")

from splinter.types import ExecutionLimits

limits = ExecutionLimits(
    max_budget=10.0,       # Stop at $10
    max_steps=100,         # Stop after 100 calls
    max_time_seconds=600,  # Stop after 10 min
)

Use with Workflow:

workflow = Workflow(workflow_id="x", limits=limits)

from splinter.types import LoopDetectionConfig

config = LoopDetectionConfig(
    max_repeated_outputs=3,  # Same output 3x = loop
    max_no_state_change=5,   # No change 5x = loop
)

workflow = Workflow(workflow_id="x", loop_detection=config)
# Raises LoopDetectedError when stuck

from splinter.control import ToolAccessController

ctrl = ToolAccessController()
ctrl.set_allowed_tools("researcher", ["web_search", "read_file"])
ctrl.set_allowed_tools("writer", ["write_file"])

ctrl.check_access("researcher", "web_search")   # ✓
ctrl.check_access("researcher", "delete_file")  # ✗ ToolAccessDeniedError

from splinter.control import RateLimiter

limiter = RateLimiter()
limiter.set_agent_limit("researcher", calls=10, window_seconds=60)
limiter.set_tool_limit("web_search", calls=20, window_seconds=60)

limiter.check_agent("researcher")        # Raises if over limit
limiter.record_agent_call("researcher")  # Record the call

from splinter.control import CircuitBreaker, CircuitBreakerConfig

breaker = CircuitBreaker(
    breaker_id="openai",
    config=CircuitBreakerConfig(
        failure_threshold=5,  # Open after 5 fails
        timeout_seconds=60,   # Retry after 60s
    )
)

breaker.check()  # Raises CircuitOpenError if open

try:
    result = await call()
    breaker.record_success()
except:
    breaker.record_failure()

Global emergency stop:

from splinter.control import CircuitBreakerRegistry

registry = CircuitBreakerRegistry()
registry.register("openai", config)
registry.register("anthropic", config)
registry.trip_all("Emergency!")  # Stop everything

from splinter.control import DecisionEnforcer, DecisionType

enforcer = DecisionEnforcer(auto_lock=True)

enforcer.record_decision(
    decision_id="strategy",
    agent_id="planner",
    decision_type=DecisionType.STRATEGY,
    value="parallel",
)

# Can't change now!
enforcer.record_decision(
    decision_id="strategy",
    agent_id="planner",
    decision_type=DecisionType.STRATEGY,
    value="sequential",  # Raises DecisionLockError
)

from splinter.control import RetryStrategy, RetryConfig, RetryMode

strategy = RetryStrategy(RetryConfig(
    max_attempts=3,
    initial_delay=1.0,
    max_delay=30.0,
    backoff_multiplier=2.0,
    mode=RetryMode.BASIC,  # or FAIL_CLOSED
))

result = await strategy.execute(unreliable_fn, arg1, arg2)

from splinter.control import RulesEngine, Rule, RuleAction

engine = RulesEngine()

engine.add_rule(Rule(
    rule_id="block_expensive",
    condition=lambda ctx: ctx.get("cost", 0) > 10,
    action=RuleAction.BLOCK,
    message="Cost > $10",
))

engine.add_rule(Rule(
    rule_id="warn_delete",
    condition=lambda ctx: ctx.get("tool") == "delete_file",
    action=RuleAction.WARN,
    message="Deleting files",
))

engine.evaluate({"cost": 15})  # Raises RuleViolationError

from splinter.control import MemoryStore

store = MemoryStore(
    max_size_bytes=10*1024*1024,  # 10 MB
    max_entries=1000,
)

store.set("researcher", "key", "value")
value = store.get("researcher", "key")
# Auto-evicts old entries when full

from splinter.coordination import SharedState

state = SharedState(initial_data={"topic": "AI"})

state.set("research.findings", ["a", "b", "c"])
findings = state.get("research.findings")

print(state.version)  # Increments on change

# Snapshot & restore
snapshot = state.snapshot()
state.set("oops", "mistake")
state.restore(snapshot)  # Rolled back

from splinter.coordination import StateOwnership

ownership = StateOwnership()
ownership.register_ownership("researcher", ["research.*"])
ownership.register_ownership("writer", ["content.*"])

ownership.check_write_permission("researcher", "research.x")  # ✓
ownership.check_write_permission("researcher", "content.x")   # ✗

from splinter.coordination import CheckpointManager, FileCheckpointStorage

mgr = CheckpointManager(storage=FileCheckpointStorage("./checkpoints"))

# Save
mgr.create_checkpoint(
    workflow_id="wf-1",
    step=3,
    agent_id="researcher",
    status=AgentStatus.COMPLETED,
    state=state,
    metrics=metrics,
)

# Resume after crash
cp = mgr.get_latest_checkpoint("wf-1")
resume_from = cp.state

from splinter.coordination import SchemaValidator

validator = SchemaValidator()

schema = {
    "type": "object",
    "properties": {"findings": {"type": "array"}},
    "required": ["findings"],
}

validator.validate(output, schema)  # Raises SchemaValidationError

from splinter.coordination import HandoffManager

handoff = HandoffManager(mode="strict")
handoff.register_schema("researcher", "writer", schema)
handoff.validate_handoff("researcher", "writer", data)

from splinter.coordination import ChainContext

ctx = ChainContext()
ctx.register_agent("researcher", "Researches topics")
ctx.register_agent("writer", "Writes articles")

ctx.record_execution("researcher", {"in": "..."}, {"out": "..."})

# Writer sees what researcher did
writer_ctx = ctx.get_context_for_agent("writer")

from splinter.coordination import GoalTracker, Goal

tracker = GoalTracker()

tracker.set_goal(Goal(
    goal_id="article",
    description="Write blog post",
    success_criteria=["Research", "Draft", "Review"],
))

tracker.mark_criterion_met("article", "Research")
tracker.update_progress("article", 0.33)

if tracker.is_achieved("article"):
    print("Done!")

from splinter.coordination import ActionEligibility

elig = ActionEligibility()
elig.set_eligible("researcher")

elig.can_act("researcher")  # True
elig.can_act("writer")      # False

elig.transfer("researcher", "writer")
elig.can_act("writer")      # True

from splinter.coordination import CompletionTracker

tracker = CompletionTracker()
tracker.require_completion("researcher")
tracker.require_completion("writer")

tracker.declare_complete("researcher", output={...})
tracker.declare_complete("writer", output={...})

if tracker.all_complete():
    print("Workflow done!")

from splinter.coordination import WaitTracker, WaitReason

tracker = WaitTracker()
tracker.start_waiting("writer", WaitReason.WAITING_FOR_INPUT, "researcher")

reason, source = tracker.get_waiting_for("writer")
# WAITING_FOR_INPUT, "researcher"

tracker.stop_waiting("writer")