Sylo SDK — Documentation

Installation

Sylo requires Python 3.10 or higher. Install from PyPI:

bash

pip install sylo-sdk

For Redis storage support:

bash

pip install "sylo-sdk[redis]"

Verify the install:

bash

python -c "import sylo; print(sylo.__version__)"
sylo --help

100% E2E Verified: Sylo runs fully locally out of the box with zero required API keys. Every feature and framework adapter is verified against live LLM infrastructure across 156 automated tests and 36 E2E integration checks.

Quickstart

Add Sylo to any async Python pipeline in three steps.

STEP 1

Initialize

python

import sylo

sylo.init(project="my-pipeline")

Runs in local mode by default. Checkpoints saved to ~/.sylo/ on disk. No external dependencies required.

STEP 2

Decorate your steps

python

import asyncio
import sylo

sylo.init(project="my-pipeline")

@sylo.step("fetch-data")
async def fetch_data(ctx: sylo.Context) -> dict:
    # Simulate or use a real LLM call
    ctx.record_token_usage(
        prompt_tokens=850,
        completion_tokens=200,
        model="gpt-4o"
    )
    return {"summary": "quarterly revenue up 12%"}

@sylo.step("analyze")
async def analyze(ctx: sylo.Context) -> dict:
    data = ctx.previous_outputs["fetch-data"]
    return {"insight": f"Analysis: {data['summary']}"}

async def main():
    async with sylo.pipeline("my-pipeline") as pipe:
        await fetch_data(pipe.context)
        await analyze(pipe.context)

asyncio.run(main())

STEP 3

Run it — then crash it

Run the pipeline once. Then add a deliberate crash inside analyze() and run it again. Sylo skips fetch-data (already checkpointed) and resumes from analyze. The terminal output shows exactly how many tokens and dollars were saved.

The demo.py in the GitHub repo shows this end to end. Run it with: python examples/demo.py

Pipelines

Every Sylo execution starts with a pipeline context manager.

python

async with sylo.pipeline("pipeline-name") as pipe:
    await my_step(pipe.context)

Each run generates a unique execution_id. Sylo uses this to track checkpoints and audit events across the entire run.

Parameter	Type	Description
"pipeline-name"	str	Identifies this pipeline in storage and logs
version	str (optional)	Defaults to "1.0". Used to isolate checkpoints between versions.
environment	str (optional)	Overrides the global environment for this run.

If a previous run of this pipeline failed, Sylo automatically detects it and offers to resume from the last successful checkpoint.

Steps & Checkpointing

The @sylo.step decorator is the core primitive. It wraps your function with checkpoint save/load logic.

python

@sylo.step("step-name")
async def my_step(ctx: sylo.Context) -> dict:
    ...

On first run: executes the function, saves output as a checkpoint. On retry: detects the existing checkpoint, loads the output, and skips re-execution. The pipeline continues as if the function ran again.

Retry configuration

python

@sylo.step("fetch-data", max_retries=3, retry_delay=2.0)
async def fetch_data(ctx: sylo.Context) -> dict:
    ...

Parameter	Default	Description
max_retries	0	Number of retry attempts on failure before raising
retry_delay	1.0	Seconds between retries (exponential backoff)

Accessing previous step outputs

python

@sylo.step("analyze")
async def analyze(ctx: sylo.Context) -> dict:
    previous = ctx.previous_outputs["fetch-data"]
    return {"result": previous["summary"]}

Recording token usage

python

ctx.record_token_usage(
    prompt_tokens=850,
    completion_tokens=200,
    model="gpt-4o"
)

In production, extract these numbers from your LLM provider's response object. In development and CI, report them manually to simulate and benchmark costs without spending real API credits.

Trust Enforcement

Every step can declare exactly what resources it is permitted to access. Sylo enforces this at runtime — if a step tries to access anything outside its declared scope, it raises SyloPermissionError immediately.

python

@sylo.step("read-emails")
@sylo.trust(
    can_read=["gmail.messages", "gmail.labels"],
    can_write=[],
    can_execute=[]
)
async def read_emails(ctx: sylo.Context) -> list:
    emails = await ctx.access(
        "gmail.messages",
        action="read",
        handler=lambda: fetch_emails_from_api()
    )
    return emails

Resources are declared as service.resource strings. Prefix wildcards are supported: gmail.* allows access to any Gmail resource. The global wildcard * is permitted but logs a warning in development mode.

Sylo logs a warning in production mode when a step runs without a @sylo.trust declaration: 'Step X has no trust declaration. Running without enforcement.'

Least privilege warnings

In development mode, Sylo warns when a step declares permissions it never uses — helping you keep declarations tight.

Approval Gates

Mark any step as requiring explicit human sign-off before execution. The pipeline pauses automatically and waits.

python

@sylo.step("delete-records")
@sylo.requires_approval(
    title="Delete customer records",
    description="Permanently delete {record_count} records",
    action_class="destructive",
    timeout_hours=24,
    on_timeout="abort",
    notify=["email", "slack"]
)
async def delete_records(ctx: sylo.Context) -> dict:
    ...

action_class	Description
destructive	Database deletes, file removal, account termination
financial	Payments, refunds, transfers
external	Emails, webhooks, third-party API writes
custom	Any user-defined class

on_timeout	Description
abort	Pipeline fails cleanly with SyloApprovalRejectedError
auto_approve	Pipeline continues automatically (use with caution)
escalate	(Coming soon) Notifies a secondary approver

Development mode

With no cloud configured, Sylo starts a local HTTP server on port 7749 and prints approve/reject URLs to the terminal:

terminal

⏸ Sylo Approval Required Pipeline: customer-operations Step: delete-records Action: Permanently delete 847 records (DESTRUCTIVE) Approve: http://localhost:7749/approve/abc123 Reject: http://localhost:7749/reject/abc123 Expires in: 24 hours | Waiting...

Click the approve URL in your browser. The pipeline resumes immediately.

Programmatic Approvals

You can also approve or reject requests programmatically from external backend webhooks or CLI scripts:

python

await sylo.approve(approval_id, decided_by="supervisor")

Storage Backends

Sylo supports three storage backends. Switch with the storage parameter on sylo.init().

Local Storage (default)

python

sylo.init(project="my-app", storage="local")

Checkpoints and audit logs saved as JSON files in ~/.sylo/. Zero dependencies. Perfect for development and single-process production deployments.

Redis

python

sylo.init(
    project="my-app",
    storage="redis",
    redis_url="redis://localhost:6379"
)

Checkpoints stored in Redis. Audit logs written to Redis Streams. Recommended for production. Requires redis package:

bash

pip install "sylo-sdk[redis]"

Sylo Cloud Coming Soon

python

sylo.init(
    project="my-app",
    storage="cloud",
    api_key="sylo_live_xxx"
)

Managed cloud storage with a dashboard, team access, and approval UI. Join the waitlist at index.html waitlist section.

Environment Variables

Configure Sylo at application startup or via standard environment variables:

Variable	Default Value	Description
SYLO_PROJECT	"default"	Identifier grouping related pipelines and agent workflows.
SYLO_ENVIRONMENT	"development"	Runtime target (`development`, `staging`, `production`).
SYLO_STORAGE	"local"	Storage driver persistence selection (`local`, `redis`, `cloud`).
SYLO_API_KEY	None	API key when communicating with Sylo Cloud.
SYLO_REDIS_URL	"redis://localhost:6379"	Connection string when using Redis backend driver.

Framework Integrations

Sylo wraps around your existing agent code without replacing your framework. Every adapter has been verified in production against real LLM infrastructure (100% E2E tested).

Framework	Adapter / Class	Verified Against	Status
LangGraph	SyloGraph	Live Groq (openai/gpt-oss-20b)	✅ 100% Tested
OpenAI Agents SDK	wrap_agent	Live Groq (openai/gpt-oss-20b)	✅ 100% Tested
CrewAI	SyloCrew	Live Groq (groq/openai/gpt-oss-20b)	✅ 100% Tested
Vanilla Python	@sylo.step, @sylo.trust	Async/Sync Python pipelines	✅ 100% Tested

1. LangGraph Integration

Wrap any LangGraph StateGraph with SyloGraph to automatically checkpoint individual nodes and skip completed steps upon resume:

python

from langgraph.graph import StateGraph, START, END
from sylo.integrations.langgraph import SyloGraph
import sylo

def research_node(state: dict) -> dict:
    return {"findings": "..."}

base_graph = StateGraph(dict)
base_graph.add_node("research", research_node)
base_graph.add_edge(START, "research")
base_graph.add_edge("research", END)

# Wrap with SyloGraph
graph = SyloGraph(base_graph, pipeline_name="langgraph-researcher")
app = graph.compile()

async def run():
    async with sylo.pipeline("langgraph-researcher"):
        return app.invoke({"topic": "quantum computing"})

2. OpenAI Agents SDK Integration

Wrap standard OpenAI Agent instances with wrap_agent (WrappedAgent). Sylo intercepts execution, records token usage automatically from Runner.run(), and caches step outputs:

python

from openai import AsyncOpenAI
from agents import Agent, Runner, set_tracing_disabled
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
from sylo.integrations.openai_agents import wrap_agent
import sylo

# Disable Agents SDK telemetry when pointing at non-OpenAI endpoints (e.g. Groq)
set_tracing_disabled(True)

client = AsyncOpenAI(base_url="https://api.groq.com/openai/v1", api_key="...")
model = OpenAIChatCompletionsModel(model="openai/gpt-oss-20b", openai_client=client)

agent = Agent(name="Researcher", instructions="Analyze breakthroughs.", model=model)
wrapped = wrap_agent(agent, step_name="research-step")

async def run():
    async with sylo.pipeline("openai-agents-pipeline"):
        return await wrapped.run("Analyze recent quantum breakthroughs.")

3. CrewAI Integration

Wrap your agents and tasks with SyloCrew. Sylo executes tasks as isolated mini-crews asynchronously inside thread pools, allowing fine-grained checkpoint recovery at the task level:

python

import litellm
litellm.drop_params = True  # Required for non-native providers (e.g., Groq)

from crewai import Agent, Task
from sylo.integrations.crewai import SyloCrew
import sylo

researcher = Agent(role="Researcher", goal="Research topic", backstory="Expert researcher", llm="groq/openai/gpt-oss-20b")
task1 = Task(description="Research quantum computing", agent=researcher, expected_output="Facts summary")

crew = SyloCrew(agents=[researcher], tasks=[task1])

async def run():
    async with sylo.pipeline("crewai-pipeline"):
        return await crew.kickoff_async()

Known Limitations & Compatibility Notes

CrewAI & Groq / LiteLLM Compatibility: When using third-party OpenAI-compatible endpoints like Groq with CrewAI, Sylo automatically patches crewai.llms.cache.mark_cache_breakpoint and strips unsupported prompt-caching markers to prevent 400 Bad Request API errors.
OpenAI Agents SDK Tracing: When pointing OpenAIChatCompletionsModel at non-OpenAI endpoints (such as Groq or local vLLM instances), make sure to call agents.set_tracing_disabled(True) to prevent trace telemetry phoning home.
Async Execution: Both OpenAI Agents SDK and CrewAI run synchronously by default. Sylo wraps them seamlessly inside asyncio.to_thread executors so they integrate natively into non-blocking async Sylo pipelines.

CLI Reference

The sylo CLI lets you inspect and replay past executions.

bash

# List recent executions for a pipeline
sylo executions list --pipeline my-pipeline --limit 20

bash

# Inspect a specific execution
sylo executions inspect <execution-id>

bash

# Replay a failed execution from a specific step
sylo executions replay <execution-id> --from-step step-name

bash

# Pretty-print the full audit log
sylo audit <execution-id>

Error Codes

Error	When it's raised	How to fix
SyloPermissionError	A step accessed a resource not in its @sylo.trust declaration	Add the resource to can_read / can_write / can_execute, or remove the access call
SyloApprovalRejectedError	An approval gate was rejected or timed out with on_timeout="abort"	Handle the exception or restructure the pipeline to avoid the rejected action
SyloStorageError	Storage operation failed in production mode	Check Redis connectivity or local disk permissions. In development mode, storage failures log warnings and never crash.
SyloCheckpointExpiredError	A checkpoint was found but has exceeded its TTL	Re-run from the beginning or increase checkpoint TTL in config
SyloConfigError	SDK used before sylo.init() was called, or invalid config	Call sylo.init() at the top of your entry point before any other Sylo calls

Getting Help

GitHub Issues

Found a bug or have a feature request? Open an issue on GitHub. Please include your Sylo version (sylo --version) and Python version in bug reports.

Open an issue ↗

Discord

Join the Sylo Discord for questions, show-and-tell, and early access discussion.

Join Discord ↗