shardflux
Python client for Shardflux: cloud computers for AI agents.
Open a persistent workspace by key, run commands in it, read and write its files, suspend it when idle, resume it later with its disk and memory intact, and fork it.
Early access. Shardflux is in early access. The API is versioned (
/v1), but this client is below 1.0: a minor release may contain breaking changes.
Versions. This README describes 0.3.0. Anything marked (0.2.0+) is not in 0.1.0, (0.2.1+) not in 0.2.0, and (0.3.0+) not in 0.2.1; the changelog (
CHANGELOG.md, shipped in the source distribution and in the wheel's metadata) lists what each version added. Check yours withpython -c "import shardflux; print(shardflux.__version__)".
- Python 3.10 or later; one dependency (
httpx), plus PyYAML with theyamlextra fortemplate.yaml. - Typed (
py.typed). - Retries, idempotency keys, operation waits and tool-token refresh are handled for you.
Install
pip install shardflux
pip install 'shardflux[yaml]' # also reads template.yaml (templates.build_from_file, 0.3.0+)
Tool-call capture integrations have extras (0.3.0+): shardflux[claude-agent-sdk],
shardflux[openai-agents], shardflux[langchain], shardflux[pydantic-ai], shardflux[crewai].
Quick start
Create a project API key in the Shardflux console (sfk_<key id>_<secret>). API keys are server
credentials; keep them out of client-side code.
from shardflux import Shardflux, format_timing
sf = Shardflux() # reads SHARDFLUX_API_KEY; or Shardflux(api_key="...")
def open_workspace():
return sf.open(key="customer-42/main", template="python-node-browser")
# 1. Open the workspace (created on first use) and run a command. Check that it worked.
ws = open_workspace()
result = ws.exec("python3 -c 'print(40 + 2)'")
if result.exit_code != 0:
raise RuntimeError(f"python3 exited {result.exit_code}: {result.stderr}")
print(result.stdout.strip()) # 42
# 2. Write a file, then suspend the workspace and wait until the suspend has finished.
ws.files.write("/home/user/notes.txt", "hello from Python\n")
ws.suspend(wait=True) # returns once suspended: ws.state == "suspended"
# 3. Open the same key again: the workspace resumes, and the file is still there.
again = open_workspace()
print(again.files.read_text("/home/user/notes.txt")) # hello from Python
print(format_timing(again.last_timing)) # (0.2.0+) where the resume's time went
open() waits until the workspace is running. Opening the same key again never resets it: files,
installed packages and running processes are still there. The same program is in
examples/quickstart.py (SHARDFLUX_API_KEY=sfk_... python examples/quickstart.py).
With 0.1.0, drop format_timing and last_timing; everything else above works as shown.
Configuration
| Argument | Environment variable | Default |
|---|---|---|
api_key |
SHARDFLUX_API_KEY |
required |
base_url |
SHARDFLUX_API_URL |
https://api.shardflux.dev |
timeout |
30.0 seconds per request |
|
max_retries |
2 (safe or idempotent requests only) |
|
http_client |
a new httpx.Client (pass your own for proxies or custom transports) |
|
on_progress (0.2.0+) |
none: a listener for the progress of every traced call (see "Timing and progress") |
Shardflux is a context manager (with Shardflux() as sf: ...); close() closes the HTTP client
it created.
Commands
r = ws.exec("pip install requests && python3 app.py", cwd="/home/user/project", env={"DEBUG": "1"}, timeout=600)
r = ws.exec(["python3", "-V"]) # a list runs as argv, without a shell
r.exit_code, r.stdout, r.stderr, r.timed_out, r.ok
A string runs through bash -lc; a list runs as argv. timeout (seconds) is enforced inside the
workspace. Pass on_output=lambda stream, chunk: ... to receive output as it arrives. If the
connection drops, exec resumes the output from byte offsets; it never starts the command twice.
Ctrl-C cancels the command in the workspace.
Files
ws.files.write("/home/user/data.bin", b"\x00\x01\x02", create_parents=True)
data = ws.files.read("/home/user/data.bin") # bytes, the whole file
text = ws.files.read_text("/home/user/notes.txt")
ws.files.list("/home/user") # {"entries": [...], "truncated": False}
ws.files.stat("/home/user/notes.txt")
ws.files.remove("/home/user/data.bin")
Writes are atomic and durable: they are acknowledged after the file and its directory are fsynced.
Lifecycle
ws.suspend(wait=True) # memory and processes are checkpointed; returns once suspended
ws.resume(wait=True) # or simply open() the key again
copy = ws.fork("customer-42/experiment") # waits until the fork is running
copy.delete() # tool access ends at once; keys are never reused
for w in sf.workspaces.list_all(key_prefix="customer-42/"):
print(w.key, w.state)
page = sf.workspaces.list(limit=50) # page.data, page.next_cursor
Requested or finished
suspend, resume, snapshot, delete, close and reset start a lifecycle operation and
return it. What the call means depends on wait:
| Call | Returns when | Returns |
|---|---|---|
ws.suspend() |
the suspend is requested (usually queued; the workspace is still running) |
the Operation |
ws.suspend(wait=True) |
the suspend has finished (ws.state is then suspended) |
the succeeded Operation |
fork() works the other way round: it waits by default and returns the new workspace; pass
wait=False to get it as soon as the fork is requested. The same calls on sf.workspaces
(sf.workspaces.suspend(workspace_id, wait=True), ...) take wait too (0.2.0+).
A failed operation raises OperationFailedError. If timeout (default 300 s) passes first,
OperationTimeoutError is raised and the operation keeps running server side: wait again with
sf.workspaces.wait_for_operation(err.operation_id). Without wait, the returned operation is the
handle for the work in progress: pass its id to wait_for_operation() when you need it finished.
A start waits for capacity for at most 15 minutes. An open, resume, restore or fork that no
host can admit yet waits in capacity_pending. Its error["details"]["deadline_at"] says when it
gives up; the phase progress event carries it as deadline_at (0.2.1+), and so does
OperationTimeoutError when your wait ends first. A start still pending at the deadline fails
with capacity_unavailable: nothing was started, the concurrency slot is released, and a
suspended workspace stays suspended with its state. OperationFailedError.retryable
(0.2.1+) is True for it, so you can tell "retry later" from a definitive failure. The client
does not retry it for you.
from shardflux import OperationFailedError
try:
ws.resume(wait=True)
except OperationFailedError as err:
if not err.retryable:
raise
# err.error_code == "capacity_unavailable": no host had room; nothing changed. Try again later.
Waiting asks the API to hold each poll until the operation changes (Prefer: wait, at most 20 s
per request), so completion arrives within one round trip of the commit (0.2.0+). Against an
API without bounded waits (or with server_wait=False) it polls with backoff (250 ms doubling to
5 s, ±20 % jitter). A waiting open() asks the API to hold the open itself until the workspace is
ready; the answer carries the first tool token, so the first tool call starts at once
(0.2.0+).
Suspended workspaces wake on use (0.2.0+)
A tool call (exec, files, changes) on a suspended workspace resumes it (or joins the resume or
open already running), then runs. A call made during a suspend or resume waits for the transition
to finish. The call never runs twice: the workspace executes nothing it refused.
- The wait is bounded per call by
transition_timeout(default 120 s), shared by the transition waits and at most 3 wakes. A resume or open still pending at the end raisesOperationTimeoutError, naming the operation, its state and reason. A failed resume or open raisesOperationFailedErrorat once. - Following an exec's output never wakes a workspace, so an explicit
suspend()is respected. ws.wake(timeout=...)does the same on demand (True when it resumed or waited, False when the workspace was already running).ws.cell(wake=None)returnsworkspace_not_runninginstead.
cell = ws.cell(transition_timeout=30) # give up waking after 30 s
cell.exec_run(["make", "test"]) # resumes the workspace first if it is suspended
Timing and progress (0.2.0+)
Every open, wake and lifecycle call is traced. ws.last_timing (and err.timing when the call
fails) says where the time went, and format_timing() prints it:
from shardflux import format_timing
ws = sf.open(key="customer-42/main", template="python-node-browser")
print(format_timing(ws.last_timing))
A slow open (34 s instead of the usual second) then reads, for example:
open 34.18 s, succeeded (workspace 01a0e5a8-3edd-74ba-b489-d62b8925e342, operation 01a0e5a8-3ef0-7ecb-975e-dff2d5ca6e33)
client: request 20.01 s (held) → capacity_pending 13.52 s (no_ready_host) → running 590 ms → view 42 ms ∥ token 61 ms
server: queued 33.40 s, ran 620 ms, total 34.02 s; start warm, boot to ready 79 ms
outside the server: 160 ms
Here the time went to waiting for a host with capacity; the network and the VM start were fast.
- client phases are on your monotonic clock: the request (a held open waits on the server:
held), then each operation state the SDK observed while waiting, with the server's reason (capacity_pending/no_ready_host: waiting for a host;running/template_downloading: fetching the template), then reading the workspace and issuing the first tool token (together:∥). - server timing comes from the operation itself (one database clock):
queuedis creation until it began running, including any wait for capacity;ranis the cell's work (placement, boot or restore, guest readiness).start/resume fromandboot to ready/host ...are what the cell reported. - outside the server is your total minus the operation's: network, TLS, polling latency, view and token. A large value with a small server total points at the connection between you and the API, not at the workspace.
- retries lists transient failures the SDK retried (cause and backoff).
The timing is a LifecycleTiming dataclass (phases, retries, server, outside_server_ms,
...). Watch it live with on_progress, on the client for every call or per call
(sf.open(..., on_progress=...), ws.suspend(wait=True, on_progress=...),
sf.workspaces.wait_for_operation(..., on_progress=...), ws.wake(on_progress=...),
ws.cell(on_progress=...)):
import sys
from shardflux import ProgressEvent, Shardflux, format_timing
def progress(e: ProgressEvent) -> None:
if e.type == "phase":
print(f"{e.action}: {e.phase}{f' ({e.reason})' if e.reason else ''} at {e.at_ms} ms", file=sys.stderr)
elif e.type == "retry" and e.retry:
print(f"{e.action}: retry {e.retry.request}: {e.retry.cause}", file=sys.stderr)
elif e.type == "done" and e.timing:
print(format_timing(e.timing), file=sys.stderr)
sf = Shardflux(on_progress=progress)
Events are phase (a phase began), retry and done (with the timing). Tool calls add token
traces (a tool token fetched: initial, expiring or invalidated) and tool events (a busy
wait, a stale token replaced, a retry). A listener that raises never breaks the call. queued and
ran need an API that reports the operation's started_at (Operation.started_at); otherwise
they are missing.
Sessions, reset and changes (0.2.0+)
# A session workspace is discarded when the session ends: close(), or the idle timeout
# (idle_timeout_seconds, default 600). The same key then opens a new, empty workspace.
with sf.open(key="agent-7/task", template="python-node-browser", lifetime="session") as ws:
ws.exec("python3 run.py")
# Leaving the block called ws.close(). For a persistent workspace close() makes no API call: it only
# drops cached tool tokens, so it is safe in `finally`.
sf.workspaces.list(lifetime="any", purpose="any", include_deleted=True) # default: persistent standard only
ws = sf.workspaces.get_by_key("customer-42/main") # any lifetime/purpose; live row over tombstones; None if absent
ws.reset(wait=True) # layered workspaces: wipe every change, back to the template
page = ws.changes(path_prefix="/home/user", hash=True, summary=True) # what changed vs. the template
for c in page.data:
print(c.change, c.path) # added, modified, metadata, deleted, replaced
Workspace views also carry disk_layout, purpose, origin, ended_reason, dev_template_id
and update_policy. Reset, changes() and save-as-template need a layered workspace.
Templates (0.2.0+)
sf.templates.files("python-node-browser", 3, "/home/user") # one directory level of a version
sf.templates.file_entry("python-node-browser", 3, "/usr/bin/python3")
sf.templates.diff("my-agent", from_="base", to=2) # .data: path, change, before, after; .summary
result = ws.save_as_template("my-agent", description="Agent with tools preinstalled")
result.build_id, result.build["target_version"]
draft = sf.templates.draft("my-agent") # dev mode: edit a template live
d = draft.create(base="python-node-browser@3")
d.workspace.exec("pip install -r requirements.txt")
state = draft.capture_state(label="deps", wait=True).result["checkpoint_id"]
with draft.open_test_instance(state_id=state) as test: # disposable copy of that state
test.exec("python3 -m pytest")
draft.publish(state_id=state, description="deps") # the next version
draft.discard()
Drafts and test instances are for owners, admins and API keys with a tool permission. A draft is a
persistent workspace (purpose="template_draft"); test instances are sessions.
Build a template from template.yaml (0.3.0+)
A template can be built from a recipe: a base, languages, packages, files, build steps and the settings every
workspace of it gets (environment, open-time inputs, start commands, services, defaults). template.yaml is that
recipe (recipe v2) in YAML. Reading YAML needs PyYAML:
pip install 'shardflux[yaml]'
(A template.json, or a template.yaml written as JSON, needs nothing; parse_yaml= accepts any parser.)
# template.yaml (schema: shardflux.template-recipe.v2 is added when absent)
base: python-node-browser@7 # services need a base whose agent runs them (python-node-browser 7+)
build:
languages:
- id: go
packages:
apt: [jq]
pip:
requirements: [/home/user/app/requirements.txt]
files:
- from: app # a local folder, relative to this file: uploaded as a tar
to: /home/user/app
owner: user
- from: config/settings.toml # a local file, uploaded byte for byte
to: /home/user/.config/app/settings.toml
owner: user
mode: "0600" # quote modes: YAML reads 0600 as a number
steps:
- name: warm-cache
run: python -c 'import app' || true
user: user
cwd: /home/user/app
settings:
env:
APP_ENV: development
inputs:
PROJECT_NAME: {kind: text, default: demo, description: Shown in the title.}
OPENAI_API_KEY: {kind: secret, required: false}
start:
- name: seed
when: create
run: python seed.py --project "$PROJECT_NAME"
cwd: /home/user/app
services:
web:
run: python -m http.server 8000
cwd: /home/user/app
ready: {port: 8000}
result = sf.templates.build_from_file("template.yaml", template_slug="my-agent", wait=True)
result.build["state"], result.build["registration"]["state"] # "published", "registered" once it is usable
result.build["provenance"]["recipe_sha256"] # the same inputs are the same template
for u in result.uploads: # one per local source: from, path, to, kind, sha256, size, uploaded, entries
print(u["from"], u["sha256"], "uploaded" if u["uploaded"] else "already there")
- Each
fromis packed (a folder: a reproducible, uncompressed tar, the same bytes on every machine), hashed and uploaded unless the organization already has those bytes; the recipe is sent withupload: "sha256:<hex>"instead.kinddefaults totarfor a folder andfilefor a file;kind: taron a file uploads a prepared uncompressed.taras it is. - Refused before any request, as
TemplateFileError:fromtogether withupload, a folder withkind: file, a compressed archive, sockets, FIFOs or devices in a folder, absolute symlinks or symlinks that leave the folder, more than 200,000 entries, more than 5 GiB, a schema other thanshardflux.template-recipe.v2. - The API validates the rest (422
validation_failedwithdetails["field"]anddetails["reason"], e.g.language_unavailable,platform_owned_path,invalid_settings). - Without
wait=Truethe call returns the queued build;on_progressreceivespack,uploadandbuildevents. Temporary archives are removed either way.
The pieces are available on their own:
from shardflux import pack_directory
packed = pack_directory("app", "app.tar") # PackResult(sha256, size, entries)
up = sf.templates.uploads.put("app.tar", "tar") # also bytes or a binary file object
up.ref, up.uploaded # "sha256:<hex>", False when the organization had the bytes
build = sf.templates.builds.create(
"my-agent",
{
"schema": "shardflux.template-recipe.v2",
"base": "python-node-browser@7",
"build": {"files": [{"upload": up.ref, "kind": "tar", "to": "/home/user/app", "owner": "user"}]},
"settings": {},
},
auto_publish=False,
)
build = sf.templates.builds.wait(build["id"], on_change=lambda b: print(b["state"]))
sf.templates.builds.list(template="my-agent").data
sf.templates.builds.log_url(build["id"])["url"] # presigned; a bearer credential, do not log it
sf.templates.builds.cancel(build["id"])
Builds belong to the API key's organization (read once from GET /v1/me; pass organization_id= to choose). A
refused upload raises TemplateUploadError (status, S3 code such as BadDigest); a wait that runs out raises
TemplateBuildTimeoutError with the last build (it keeps going server side).
Export, test, inputs and startup
exported = sf.templates.versions.recipe("my-agent", 3) # "Edit template": the recipe in request form
sf.templates.builds.create("my-agent", exported["recipe"]) # same base and uploads: same recipe_sha256
exported["settings"] # env, inputs, start, services, defaults as the version has them
detail = sf.templates.get("my-agent") # versions with their settings; category "os"/"stack"/None
sf.templates.list(owner="platform").data
# A throwaway session on any version (published or not), with its inputs:
with sf.templates.version_test_instances.create("my-agent", 4, inputs={"PROJECT_NAME": "try"}) as ws:
ws.exec("curl -s localhost:8000")
# Open with the template's text inputs; the version's start commands and services run at start.
ws = sf.open(key="customer-42/main", template="my-agent", inputs={"PROJECT_NAME": "acme"})
ws.inputs() # {"PROJECT_NAME": "acme"}; secret inputs bind the stored secret of the same name
ws.startup # {"state": "ready", "trigger": "create", ...}; None without start commands or services
A failed start command or service leaves the workspace running with startup["state"] == "failed" (step,
service, exit_code, output_tail, reason); the next open runs the failed step again. Inputs the version
does not declare are 422 input_unknown; a missing required one is input_required.
draft.create(display_name=..., inputs=...), draft.open_test_instance(inputs=...), draft.publish(settings=...)
and ws.save_as_template(..., settings=...) take the same settings (each field given replaces the source version's;
settings["defaults"] together with defaults is 422 invalid_settings).
Languages and packages
langs = sf.templates.languages("python-node-browser@7")
[(l["id"], l["version"], l["included"]) for l in langs["data"]] # included: the base has it already
sf.templates.packages.search("npm", "typescript")["data"] # name, version, summary
sf.templates.packages.search("apt", "ffmpeg", base="ubuntu-24.04@1") # apt searches the base's index
sf.templates.packages.get("pip", "pandas")["versions"]
Secrets (0.2.0+)
Store credentials once and give them to a workspace's processes as environment variables. Values
are write-only: no API returns them. Every exec and terminal in the workspace receives the
secrets bound to it, plus any the call names in secret_refs.
import os
project_id = sf.me()["api_key"]["project_id"]
sf.secrets.create(project_id, "OPENAI_API_KEY", os.environ["OPENAI_API_KEY"])
# Bind by name when opening (a new key gets the binding; an existing key has it replaced).
ws = sf.open(key="customer-42/main", template="python-node-browser", secrets=["OPENAI_API_KEY"])
ws.secrets.get() # {"names": [...], "secrets": [{"name", "status", "secret_id", "scope"}]}
ws.secrets.set(["OPENAI_API_KEY", "DATABASE_URL"]) # replace; [] clears
ws.exec("python3 agent.py") # sees $OPENAI_API_KEY and $DATABASE_URL
- A name that is unknown, or a secret this workspace may not use, raises
ShardfluxApiError422 (err.reason == "secret_not_available",details["names"]); nothing changes. A bound secret must allow theexecandptytools (the default). statusper bound name:available,not_allowed(its permissions no longer cover this workspace; starts are refused with 403 until fixed) ordeleted.- Deleting a secret removes it from every binding. Forks keep the binding, but secrets limited to specific workspaces are checked against the fork's own id.
sf.secretsalso haslist,get,update,rotate,versions,delete,access_events,create_organizationandlist_organization. Permission arguments you do not pass are left alone;Nonemeans "no restriction". Organization-wide secrets and access logs belong to organization owners and admins, so a project API key gets 403 for those.
Tool-call capture (0.3.0+)
Your harness runs the model loop and your own tools (search, SQL, HTTP APIs, MCP servers). Their results reach
the model's context but not the workspace. Tool-call capture saves every call's input and output as files in the
workspace, so the agent can work on them with jq or pandas. They are also kept in snapshots and forks.
capture = ws.capture_tool_calls() # one run directory under /home/user/tool-calls
Capture is invisible to the harness:
- A wrapped tool returns the same object and raises the same exception. A sync tool stays sync, and an async tool stays async.
- Writes happen on background threads. A call costs one snapshot of its data on the calling thread, well under a millisecond for typical results.
- Nothing raised by capture reaches your code after
capture_tool_calls(). Write failures, overflow and serialization problems go toon_error(aCaptureError) and tocapture.stats. - The model's view is never changed.
Pick the pattern that matches your harness.
Hand-rolled loop (Anthropic, OpenAI). Record each call with the model's id:
for block in response.content:
if block.type == "tool_use":
with capture.call(block.name, block.input, call_id=block.id) as call:
call.output = my_tools[block.name](**block.input)
# or afterwards: capture.record(block.name, block.input, output, call_id=block.id)
Decorated tools (anthropic @beta_tool/@beta_async_tool, openai-agents @function_tool, pydantic-ai
@agent.tool/@agent.tool_plain, LangChain @tool, CrewAI @tool, llama-index FunctionTool.from_defaults).
Stack @capture_tool directly under the framework's decorator. The framework still sees the same signature,
docstring, type hints (including from __future__ import annotations) and async-ness, so the model gets the
same schema.
from shardflux import capture_tool
@function_tool
@capture_tool # records to the capture active in this context; passes through when there is none
def search_orders(ctx: RunContextWrapper, customer: str) -> list[dict]: ...
with capture.activate(): # per request/conversation, e.g. in a multi-tenant server
await Runner.run(agent, "...")
@capture.tool(orcapture.wrap(fn)) binds a tool to one capture instead.- The input is the call's arguments by name. Injected framework parameters are left out:
RunContext,ToolContext,RunnableConfig,ToolRuntime, callback managers, and atool_call_idparameter. The call id is read from them.exclude_args=("password",)leaves out more. - smolagents warns about extra decorators and rejects async tools, so use the call form:
tool(capture_tool(fn)). - Generators are passed through item by item and stored as
.jsonl. A generator closed early is recorded asincomplete.
Claude Agent SDK (pip install 'shardflux[claude-agent-sdk]'):
from shardflux.integrations.claude_agent_sdk import capture_hooks
options = ClaudeAgentOptions(hooks=capture_hooks(capture, merge=my_hooks))
PostToolUseandPostToolUseFailurerecord every tool, keyed bytool_use_id.- A
PreToolUsehook matching^mcp__shardflux__waits for pending writes before a Shardflux MCP tool runs, so that tool sees them. Other tools are not slowed.
OpenAI Agents SDK (shardflux[openai-agents]):
from shardflux.integrations.openai_agents import CaptureRunHooks
result = await Runner.run(agent, "...", hooks=CaptureRunHooks(capture, inner=my_hooks))
- Function tools are recorded with the
call_idand raw arguments fromToolContext. - Hosted tools (MCP, code interpreter, file search, web search, image generation) are recorded from
on_llm_end. - Hooks see a failing tool only as the SDK's error string, which is recorded with status
error. Add@capture_toolunder@function_toolto record the exception itself. The shared call id keeps it to one line.
LangChain / LangGraph (shardflux[langchain]):
from shardflux.integrations.langchain import CaptureCallbackHandler
agent.invoke({"messages": [...]}, config={"callbacks": [CaptureCallbackHandler(capture)]})
A ToolMessage is stored as its content, together with its artifact when it has one. status="error" is
recorded as an error.
Pydantic AI 2.x (shardflux[pydantic-ai]):
from shardflux.integrations.pydantic_ai import capture_capability
agent = Agent(model, capabilities=[capture_capability(capture)])
- Function tools and MCP toolsets are recorded with their
tool_call_id. AModelRetryis recorded asretry. - Provider-run (native) tools are recorded from the model response.
- On pydantic-ai 1.x, use the decorator path instead.
CrewAI (shardflux[crewai]). This uses CrewAI's global tool hooks. CrewAI passes no call id, so don't also
decorate the same tools.
from shardflux.integrations.crewai import register_hooks
unregister = register_hooks(capture)
crew.kickoff()
unregister()
MCP clients (any ClientSession or mcp.Client; no extra):
from shardflux.integrations.mcp import instrument
release = instrument(session, capture, server="github") # recorded as "github.<tool>"; release() undoes it
Selecting tools
- Explicit capture always records:
record,call,wrap,@capture.tooland@capture_tool. - The hook integrations record every tool, including Shardflux's own. Narrow them with
tools=/exclude=, each given as tool names, a compiled pattern or a predicate:
ws.capture_tool_calls(exclude=re.compile(r"^mcp__shardflux__"))
ws.capture_tool_calls(tools=["web_search", "sql"])
- Calls are deduplicated by call id (the last 10,000), and the first record wins. A decorator and a hook on the same tool therefore produce one line. When the decorated function doesn't receive the call id (no context parameter), the OpenAI Agents, LangChain and Pydantic AI integrations announce each call as it starts, and the wrapper takes the id of the announced call with the same tool name and matching arguments (announcements are kept for 10 minutes, at most 1,000). CrewAI and MCP pass no id to match, so don't combine those hooks with a decorator on the same tool.
transform(event) -> event | Noneredacts or drops a call. It receives a JSON copy of{tool, call_id, source, status, error, started_at, duration_ms, input, output, meta}. If it raises, the call is dropped, never written unredacted.- Nothing is redacted by default. The input came from the model, so nothing in it is secret from the agent.
Layout
/home/user/tool-calls/README.md layout and jq recipes (written once per capture)
/home/user/tool-calls/<run>/index.jsonl one JSON line per call, in completion order
/home/user/tool-calls/<run>/000001-web_search.json
/home/user/tool-calls/<run>/000002-fetch.html
/home/user/tool-calls/<run>/000003-screenshot.png
/home/user/tool-calls/<run>/000004-github.search/ multi-part output (MCP content, Anthropic blocks):
part-1.txt, part-2.png, result.json
/home/user/tool-calls/<run>/000005-sql.input.json an input over 64 KiB
/home/user/tool-calls/<run>/000006-scrape.json.part output cut at max_output_bytes ("truncated": true)
- The run id is
YYYYMMDDTHHMMSSmmmZ-xxxxxx(capture.run_id,capture.run_dir). It is always generated. - To group runs, for example per conversation, point
dirat a subdirectory:ws.capture_tool_calls(dir="/home/user/tool-calls/conv-123"). - An index line has
v, run, seq, call_id, tool, source, status(ok | error | cancelled | incomplete | retry),error, started_at, duration_ms, input(orinput_path),output_path(relative to the run directory),content_type, bytes, sha256, truncated. It can also haveparts,droppedandmeta(merged fromcapture_tool_calls(meta=...)and the call'smeta). - Read it with
jq -cR 'fromjson? // empty'so a partly written line is skipped. The workspace README has recipes.
capture.prompt_hint() returns a short paragraph telling the agent where its tool results are. Add it to your
system prompt if you want the agent to know; capture never injects it.
Reading your own writes, lifecycle, exit
- On the same client,
ws.exec,ws.files.*,ws.changes,snapshot,suspend,fork,closeandsave_as_templatefirst wait for capture writes recorded before them. The wait is bounded bysettle_timeout(30 s) and never fails the call. Traced calls show it as acapture_flushphase. deleteandresetdiscard pending writes. After a reset, the README is written again.- Writes resume a suspended workspace, like any tool call. With
wake=Falsethey never do: they retry forretry_window(120 s) and are then dropped aswrite_failed. capture.flush(timeout)/await capture.aflush(timeout)wait for everything recorded so far.close()/aclose(),withandasync withstop recording and flush.Shardflux.close()flushes every capture of the client.- At interpreter exit, one
atexithandler flushes pending writes, bounded byexit_timeout(5 s). - Serverless platforms freeze or kill the process when the handler returns, so flush before returning
(Lambda:
capture.flush(), orawait capture.aflush()in async handlers). - A forked child process (
os.fork, multiprocessing with fork) gets inert captures and oneforkederror. Open a capture in the child if it needs one.
Limits
- Outputs are capped at
max_output_bytes(32 MiB) per call:- Text and JSON over the limit are cut at a UTF-8 boundary and stored as
.part. - Binary or multi-part output over the limit is not stored (
dropped: "too_large").
- Text and JSON over the limit are cut at a UTF-8 boundary and stored as
- Pending writes are bounded, and nothing ever blocks the tool:
max_pending_bytes(128 MiB) counts everything a pending call holds: its files, its inline input and its index line. Over it, the output is dropped, and so is an input that alone doesn't fit (inputnull,meta.input_dropped: "queue_full"). A small index line withdropped: "queue_full"is kept; these lines may go past the limit.max_pending_calls(10,000) is the hard bound for a workspace that takes no writes. Past it, a call is not recorded at all andon_errorgetsqueue_full.
- Each call costs about one HTTP write, plus batched index appends (every 50 ms or 1 MiB). Up to
concurrency(4) writes of a capture are in flight at once: files of different calls go in parallel, so one large upload doesn't hold back later calls. Index lines are still appended inseqorder, each only after its files are acknowledged. More than roughly 100 calls per second per workspace reaches the pending limits. - A retryable failure (network, 5xx,
workspace_busy,rate_limited, disk full) retries for up toretry_window. After a network error the retry keeps its idempotency key, so a write that committed is not repeated. After an error response it moves to a new key, because the host replays a failed write's error to every retry with the same key. A 400, 403, 413 or 422 drops the item. A 404 or 410 closes the capture (the workspace is gone). - On template v1 the files API writes as root, so the agent can read captured files but not modify them.
on_progresslisteners of the client also receive the capture writers' tool events, from writer threads.
Errors
All errors derive from ShardfluxError.
ShardfluxApiError: the API or the workspace refused the request. It mirrors the error envelope:code,message,request_id,retryable, plusstatus,details,operation_id,retry_afterandreason(0.2.0+) (details["reason"], e.g.workspace_not_running,not_session,legacy_disk_layout).OperationFailedError: an awaited operation endedfailedorcanceled(error_code,retryable(0.2.1+),operation).retryableis the operation error's own flag:Trueforcapacity_unavailable(no host could admit the start before its deadline; retry later),Falsefor a definitive failure.OperationTimeoutError: waiting gave up; the operation continues (operation_id,last_state,last_reason,deadline_at(0.2.1+) while it waits for capacity).ShardfluxProtocolError: a response was not the documented shape.TemplateFileError,TemplateUploadError,TemplateBuildTimeoutError(0.3.0+): a template file or local path that cannot be used, bytes the storage refused, a build wait that ran out (see "Build a template from template.yaml").
Each of them has timing (0.2.0+) when a traced call (open, lifecycle call, wait, wake, token
fetch) failed with it: format_timing(err.timing) says where the time went before the failure.
from shardflux import ShardfluxApiError
try:
sf.open(key="customer-42/main", template="python-node-browser")
except ShardfluxApiError as err:
print(err.code, err.reason, err.message, err.request_id, err.retryable)
Treat unknown error codes and reasons as generic errors: show message, and use retryable.
Anything else
sf.me() returns the API key's organization and project. sf.secrets manages secrets and
sf.templates templates, builds, uploads, file trees, diffs and drafts (see above). sf.request(method, path, ...) calls
any /v1 route with the client's authentication, retries and error handling. The TypeScript SDK
(npm install @shardflux/sdk) covers the full API, including agent tool definitions for model
providers.
Compatibility
- The client follows the API's
/v1contract. New fields, enum values and error codes can appear in any release; ignore unknown fields. - While below 1.0, a breaking change bumps the minor version (0.1 to 0.2).
shardflux.__version__is exported; requests sendUser-Agent: shardflux-sdk-python/<version>.- Examples in this README and in
examples/name the version they need.
License
Apache-2.0
Metadata
Release files for shardflux 0.3.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| shardflux-0.3.0.tar.gz | 239.8 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| shardflux-0.3.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 380.4 kB