Give your agent workspace tools

Your agent loop, the workspace's tools

Your application keeps the agent loop and the model calls. The workspace is the computer the agent's tools act on. workspaceTools(workspace) from @shardflux/sdk returns the tools: each has a name, a description, a JSON Schema for its parameters and an execute function that runs the call in the workspace.

import { executeToolCall, toAnthropicTools, toOpenAITools, workspaceTools } from '@shardflux/sdk';

const tools = workspaceTools(workspace);

const anthropicTools = toAnthropicTools(tools);                       // Anthropic Messages API
const chatTools = toOpenAITools(tools);                               // OpenAI Chat Completions
const responsesTools = toOpenAITools(tools, { api: 'responses' });    // OpenAI Responses API

// For each tool call the model makes:
const output = await executeToolCall(tools, { name: call.name, input: call.input });

executeToolCall finds the tool by name, accepts the arguments as an object (input, Anthropic) or a JSON string (arguments, OpenAI), validates them against the tool's schema, and runs the tool. An Anthropic tool_use block or an OpenAI Responses function_call item can be passed as it is. (0.8.0+) The exported definitions are typed as the providers' own tool types (Anthropic.Tool[], OpenAI.Chat.ChatCompletionTool[] and, with { api: 'responses' }, OpenAI.Responses.FunctionTool[]), so they type-check under TypeScript's strict checks without casts. The workspace keeps its files, installed packages and processes between calls and between conversations. If the workspace is suspended, the first tool call resumes it.

In Python (shardflux 0.4.0+), workspace_tools(ws) returns the same tools with the same names and schemas (search_files and edit_file from 0.6.0):

from shardflux import execute_tool_call, to_anthropic_tools, to_openai_tools, workspace_tools

tools = workspace_tools(ws)

anthropic_tools = to_anthropic_tools(tools)                # Anthropic Messages API
chat_tools = to_openai_tools(tools)                        # OpenAI Chat Completions
responses_tools = to_openai_tools(tools, api="responses")  # OpenAI Responses API

# For each tool call the model makes: a tool_use block, an OpenAI tool call or function_call item, or a dict.
output = execute_tool_call(tools, block)

The Python tools are synchronous; in async code, run them with asyncio.to_thread. An agent that speaks MCP gets the same tools from the MCP server.

Tools before the workspace exists

Give the tools a key instead of a workspace (SDK 0.15.0+, Python 0.11.0+): they are built from your API key's tool permissions without starting anything, and the model's first tool call creates the workspace. Use the identifier your application already has (the user, the thread, the repository) as the key, so every conversation of that user or thread gets its own computer back.

import { executeToolCall, toAnthropicTools, workspace } from '@shardflux/sdk';

const tools = await workspace(`acme/${threadId}`, { template: 'default' }).tools();
const anthropicTools = toAnthropicTools(tools);
// For each tool call the model makes; the first one creates the workspace:
const output = await executeToolCall(tools, block);

from shardflux import execute_tool_call, to_anthropic_tools, workspace

tools = workspace(f"acme/{thread_id}", template="default").tools()
anthropic_tools = to_anthropic_tools(tools)
output = execute_tool_call(tools, block)  # the first call creates the workspace

See Workspaces by key.

A complete loop with Anthropic

npm install @shardflux/sdk @anthropic-ai/sdk
export ANTHROPIC_API_KEY=...

import Anthropic from '@anthropic-ai/sdk';
import { Shardflux, executeToolCall, toAnthropicTools, workspaceTools } from '@shardflux/sdk';

const cloud = new Shardflux({ apiKey: process.env.SHARDFLUX_API_KEY! });
const anthropic = new Anthropic(); // reads ANTHROPIC_API_KEY

const workspace = await cloud.workspaces.open({ key: 'agent-demo/main', template: 'default' });
const tools = workspaceTools(workspace, { tools: ['exec', 'files'] });

const messages: Anthropic.MessageParam[] = [
  { role: 'user', content: 'Write /home/user/fizzbuzz.py, run it for 1 to 15, and tell me what it printed.' },
];

void workspace.hint().catch(() => {}); // begin the turn while the model thinks
for (;;) {
  const response = await anthropic.messages.create({
    model: 'claude-opus-5',
    max_tokens: 16000,
    tools: toAnthropicTools(tools),
    messages,
  });
  messages.push({ role: 'assistant', content: response.content });

  if (response.stop_reason !== 'tool_use') {
    for (const block of response.content) if (block.type === 'text') console.log(block.text);
    await workspace.suspendWhenIdle({ afterSeconds: 0 });   // the turn is over: suspend as soon as idle
    break;
  }

  // Run every tool call of this turn, and send all results back in one message.
  const results: Anthropic.ToolResultBlockParam[] = [];
  for (const block of response.content) {
    if (block.type !== 'tool_use') continue;
    try {
      const output = await executeToolCall(tools, block);
      results.push({ type: 'tool_result', tool_use_id: block.id, content: JSON.stringify(output) });
    } catch (err) {
      results.push({ type: 'tool_result', tool_use_id: block.id, content: String(err), is_error: true });
    }
  }
  messages.push({ role: 'user', content: results });
}

Run it with node agent.ts (in a package with "type": "module"). Errors go back to the model as is_error results, so it can correct a bad argument or a failing command. When the model answers without a tool call, the turn is over, and the loop asks for a suspend as soon as the workspace is idle (0.10.0+); see Suspend when the turn ends.

A complete loop with OpenAI

npm install @shardflux/sdk openai
export OPENAI_API_KEY=... OPENAI_MODEL=...   # the OpenAI model you use

With the Responses API, a function_call item can be passed to executeToolCall as it is. This loop continues the conversation with previous_response_id, so each turn sends only the tool results:

import OpenAI from 'openai';
import { Shardflux, executeToolCall, toOpenAITools, workspaceTools } from '@shardflux/sdk';

const cloud = new Shardflux({ apiKey: process.env.SHARDFLUX_API_KEY! });
const openai = new OpenAI(); // reads OPENAI_API_KEY

const workspace = await cloud.workspaces.open({ key: 'agent-demo/main', template: 'default' });
const tools = workspaceTools(workspace, { tools: ['exec', 'files'] });

let input: OpenAI.Responses.ResponseInput = [
  { role: 'user', content: 'Write /home/user/fizzbuzz.py, run it for 1 to 15, and tell me what it printed.' },
];
let previousResponseId: string | undefined;

for (;;) {
  const response = await openai.responses.create({
    model: process.env.OPENAI_MODEL!,
    tools: toOpenAITools(tools, { api: 'responses' }),
    input,
    previous_response_id: previousResponseId,
  });
  const calls = response.output.filter((item) => item.type === 'function_call');
  if (calls.length === 0) {
    console.log(response.output_text);
    await workspace.suspendWhenIdle({ afterSeconds: 0 });   // the turn is over: suspend as soon as idle
    break;
  }

  previousResponseId = response.id;
  input = [];
  for (const call of calls) {
    const output = await executeToolCall(tools, call).catch((err: unknown) => ({ error: String(err) }));
    input.push({ type: 'function_call_output', call_id: call.call_id, output: JSON.stringify(output) });
  }
}

For Chat Completions, pass toOpenAITools(tools) as tools and dispatch each function call of message.tool_calls (call.type === 'function') with executeToolCall(tools, { id: call.id, name: call.function.name, arguments: call.function.arguments }). A message without tool_calls ends the turn: call workspace.suspendWhenIdle({ afterSeconds: 0 }) there.

A complete loop in Python

(shardflux 0.4.0+; suspend_when_idle 0.6.0+) The same Anthropic loop with the Python SDK:

pip install shardflux anthropic
export SHARDFLUX_API_KEY=... ANTHROPIC_API_KEY=...

import json

import anthropic
from shardflux import Shardflux, execute_tool_call, to_anthropic_tools, workspace_tools

sf = Shardflux()  # reads SHARDFLUX_API_KEY
client = anthropic.Anthropic()  # reads ANTHROPIC_API_KEY

ws = sf.open(key="agent-demo/main", template="default")
tools = workspace_tools(ws, tools=["exec", "files"])

messages = [
    {"role": "user", "content": "Write /home/user/fizzbuzz.py, run it for 1 to 15, and tell me what it printed."}
]
while True:
    response = client.messages.create(
        model="claude-opus-5", max_tokens=16000, tools=to_anthropic_tools(tools), messages=messages
    )
    messages.append({"role": "assistant", "content": response.content})
    if response.stop_reason != "tool_use":
        print("".join(block.text for block in response.content if block.type == "text"))
        ws.suspend_when_idle(after_seconds=0)  # the turn is over: suspend after a minute idle
        break
    # Run every tool call of this turn, and send all results back in one message.
    results = []
    for block in response.content:
        if block.type != "tool_use":
            continue
        try:
            output = execute_tool_call(tools, block)
            results.append({"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(output)})
        except Exception as err:  # bad arguments or a refused call: tell the model, so it can correct itself
            results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(err), "is_error": True})
    messages.append({"role": "user", "content": results})

With the OpenAI Responses API, pass to_openai_tools(tools, api="responses") as tools, pass each function_call item of response.output to execute_tool_call as it is, and send the result back as {"type": "function_call_output", "call_id": item.call_id, "output": json.dumps(output)}. When response.output has no function_call item, the turn is over: call ws.suspend_when_idle(after_seconds=0). A Chat Completions tool call (message.tool_calls[i]) can also be passed as it is.

Suspend when the turn ends

A workspace uses RAM GiB-hours while it works (parked between tool calls, little or none) until its idle timeout suspends it. Your loop knows when a turn ends, so the loops above end each turn with workspace.suspendWhenIdle({ afterSeconds }) (@shardflux/sdk 0.10.0+) or ws.suspend_when_idle(after_seconds=...) (shardflux 0.6.0+): once the workspace has been idle for afterSeconds (0 to 3600; 0 = as soon as it is idle), it is suspended.

• If the user answers within that time, the next turn's first tool call cancels the request and finds the workspace running. Otherwise the workspace suspends, and the next tool call wakes it.
• A command the agent left running (counted until it ends or its timeout passes, 1 hour when it has none), an attached output stream or a keepalive postpones the suspend until afterSeconds after it ends.
• The idle time counts from the later of the workspace's last work and the request. Calling it again replaces the request, and cancelSuspendWhenIdle() (cancel_suspend_when_idle()) cancels it.
• It never delays a suspend the idle policy would do sooner, and it applies under every policy, never included.
• Choose afterSeconds by how quickly your users usually reply: long enough to cover a quick answer, short enough to free the workspace promptly when the conversation is over.

A job that nobody is waiting on does not need the grace period: suspend it at once (see Run the agent on a schedule or an event). All the rules are in Suspend when idle.

Other agent frameworks

The tools are plain objects, so any framework that takes a name, a description, a JSON Schema and a function can use them:

import type { WorkspaceTool } from '@shardflux/sdk';

const tools: WorkspaceTool[] = workspaceTools(workspace);
for (const t of tools) console.log(t.name, t.permission, JSON.stringify(t.parameters));

const result = await tools.find((t) => t.name === 'exec')!.execute({ command: 'python3 --version' });

execute validates its arguments against the same schema the model saw and throws ToolArgumentError (with issues) when they do not match. In Python, a WorkspaceTool has the same fields; call tool.execute({"command": "python3 --version"}). Its ToolArgumentError is also a ValueError.

The tools

Which tools you get depends on the API key's tool permissions and the tools option.

Command output and file content returned to the model are cut at 64 KiB per call (truncated: true); maxOutputBytes (max_output_bytes in Python) changes that. search_files returns whole matches within that limit and counts the rest in omitted_matches. edit_file applies its edits only to the file's revision: the expected_revision the model gives (the revision of its previous edit of that file), else the revision the tool reads just before editing, so a change made in between fails with revision_mismatch instead of being overwritten. See Search and edit files. In Python, the cut never splits a UTF-8 character, and terminal_read's next_offset counts exactly the bytes it returned, so output past the limit comes with the next read.

edit_file replaces exact text: each old_text must occur exactly once in the file unless replace_all is true, and the edits apply in order, all of them or none. When the model gives no expected_revision, the tool reads the file's revision first and pins the edit to it, so a change made in between fails the edit (409 conflict, reason: revision_mismatch) instead of being overwritten. The result's revision can be passed as expected_revision to the next edit_file of the same file.

Long-running commands

(TypeScript 0.13.0+, Python 0.9.0+) On a processful workspace, exec with background: true starts the command and returns its session_id at once, so the agent keeps working with the other tools while a build, a test suite or a training run goes on. exec_read reports the command's state, exit code and output: the end of each stream, or only what is new when you pass back next_stdout_offset and next_stderr_offset. With wait_ms it waits up to that long for the command to exit and answers as soon as it does. exec_cancel stops it.

const run = (name: string, input: Record<string, unknown>) => executeToolCall(tools, { name, input });

const { session_id } = (await run('exec', {
  command: 'python3 train.py --epochs 40',
  background: true,
  timeout_ms: 3 * 3_600_000, // the longest it may run
})) as { session_id: string };
// ...the agent keeps using the other tools...
const progress = await run('exec_read', { session_id, wait_ms: 60_000 });

Give a long job a timeout_ms that covers its whole run (up to 24 hours): the workspace stays awake until the command ends or that time passes, and its idle timeout counts from there (what keeps a workspace awake). The output stays in the workspace, so a later turn reads it with the same session_id.

(TypeScript 0.9.0+, Python 0.6.0+) Each tool call first tells the workspace that a call is coming (a wake hint, sent without waiting for it), so a workspace its host parked while idle is being restored while the call is prepared. read_file, list_files and search_files send no hint, so reading a suspended workspace does not resume it through the hint. Pass hint: false (hint=False in Python) to turn it off, for example when you send workspace.hint() yourself as soon as the model starts a tool call.

Options

const tools = workspaceTools(workspace, {
  tools: ['exec', 'files', 'git'],   // default: the tools of the workspace's last token, else all six
  agentLabel: 'coder',               // attribution: one agent session per label
  prefix: 'workspace_',              // tool names become workspace_exec, workspace_read_file, ...
  maxOutputBytes: 65_536,            // bytes of output or file content returned to the model (default 64 KiB)
  defaultCwd: '/home/user/project',  // exec's working directory when the model gives none (default: the user's home)
  transitionTimeoutMs: 120_000,      // longest wait per call for a suspended workspace to wake (default 120 s)
  hint: true,                        // (0.9.0+) send a wake hint when a call starts (default true)
});

Pass wake: null to make calls on a suspended workspace fail with workspace_not_running instead of resuming it.

(0.9.0+) Each tool call first sends a wake hint without waiting for it, so a workspace its host has parked while idle starts waking while the call runs (read_file, list_files and search_files send none: a sleeping workspace answers them from its disk). hint: false turns that off, for example when you send workspace.hint() yourself as soon as the model starts writing a tool call. On a file-first workspace the tools are exec and the files tools only, and exec runs each command as an execution: its result adds execution_id, state, tree_revision and changed. mode builds the definitions without reading the workspace, and onExecution(id) is called with each execution id before it is sent. In Python the options are keyword arguments with the same meaning: workspace_tools(ws, tools=[...], agent_label=..., prefix=..., max_output_bytes=..., default_cwd=..., transition_timeout=120, wake=None, hint=True) (hint from 0.6.0).

Run the agent on a schedule or an event

Scheduled and event-driven work starts where your agent loop runs: in your application. Have the job runner or webhook handler you already use (cron, Vercel Cron, Inngest, Trigger.dev, Celery beat, GitHub Actions) call your agent, and open the customer's workspace by key as usual. A suspended workspace resumes on the open, so nothing has to stay running between runs. Your scheduler stays the source of truth, and a suspended workspace costs no compute between runs (see what keeps a workspace awake).

// Called by your job runner, once per customer.
export async function weeklyReport(customerId: string) {
  const workspace = await cloud.workspaces.open({ key: `customer/${customerId}`, template: 'python-node-browser' });
  try {
    await runAgent(workspace, 'Update the weekly report with the new files in /home/user/data.');   // your loop
  } finally {
    await workspace.suspend({ wait: true });   // stops its compute now, not at its idle timeout
  }
}

def weekly_report(customer_id: str) -> None:
    ws = sf.open(key=f"customer/{customer_id}", template="python-node-browser")
    try:
        run_agent(ws, "Update the weekly report with the new files in /home/user/data.")   # your loop
    finally:
        ws.suspend(wait=True)

When one schedule covers many customers:

• Keep concurrency below your plan's limit of workspaces working at once. Workspaces count while they work, across your organization, including the ones your users are working in. Past the limit, an open waits up to 10 seconds for a workspace to stop working and is then refused with 429 quota_exceeded (details.limit: concurrent_workspaces) and Retry-After. The SDKs send it again for up to admissionTimeoutMs / admission_timeout (2 minutes by default; TypeScript 0.22.0+, Python 0.18.0+) before they raise it. Run the jobs through a queue with a concurrency limit, for example 40 on Startup (limit 50), and let the queue retry a refused job later.
• Suspend when the job is done, with suspend(). Nobody is waiting for a reply, so end its compute at once. suspendWhenIdle is for interactive turns, where the user may answer within a minute. Otherwise each workspace suspends at its learned idle timeout (5 minutes while there is no history to learn from), and one whose processes keep running keeps working, and counting toward workspaces working at once, until then. If one of your users makes a tool call during the suspend, the call waits and then wakes the workspace again.
• Or keep only files. When a job needs nothing but its files between runs, a file-first workspace never counts toward workspaces working at once, and compute is metered only while a command runs.
• Make jobs safe to retry. Opening a key never resets its workspace, so running a job again is safe as long as your own steps are. A failed start (retryable: true) changed nothing and can simply be sent again.

See Pricing and limits for each plan's workspaces working at once and allowances.

Save your harness's tool calls into the workspace

Your own tools (web search, SQL, HTTP APIs, other MCP servers) run in your application, so their results reach the model but not the workspace. Tool-call capture (@shardflux/sdk 0.7.0+, shardflux 0.3.0+) saves each call's input and full output as files in the workspace, where the agent can work on them with jq or Python, and where snapshots and forks keep them.

const capture = workspace.captureToolCalls();
const tools = capture.tools(workspaceTools(workspace));   // Shardflux tools are recorded too

// Inside your loop, for each tool_use block of the model's reply:
const output = myTools[block.name]
  ? await capture.run(block, () => myTools[block.name]!(block.input))   // your tool: recorded, same return value
  : await executeToolCall(tools, block);

// Before your process or serverless function ends:
await capture.flush();

In Python (0.4.0+): tools = capture.tools(workspace_tools(ws)), then execute_tool_call(tools, block) records the call with block.id.

Capture is invisible to your harness: a wrapped tool returns the same value and throws the same error, and nothing capture does throws into your code (failures go to onError and capture.stats). Writes run in the background.

In the workspace, under /home/user/tool-calls/<run>/:

index.jsonl                 one JSON line per call: seq, call_id, tool, status, input, output_path, ...
000007-web_search.json      a call's output (.json, .txt, .html, .png, .pdf, ... from its content)
000010-github.search/       an MCP result or content blocks: part-1.txt, part-2.png, result.json

capture.promptHint() returns a paragraph telling the agent where its tool calls are; add it to your system prompt if you want. The agent can then run, for example, jq -cR 'fromjson? // empty' /home/user/tool-calls/*/index.jsonl.

A Python hand-rolled loop:

capture = ws.capture_tool_calls()

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)

capture.flush()

• Redaction. Pass transform to redact, change or drop a call before it is written; if it throws, the call is dropped, never written unredacted.
• Read your own writes. Exec and file calls through the same client, and suspend, snapshot and fork, first wait (up to 30 seconds) for captured writes recorded before them.
• Limits. An output over 32 MiB is cut (text and JSON) or not stored (binary). A tool call never waits for capture, and the default pending limits handle about 100 calls per second per workspace.
• Serverless. Flush before the function returns: waitUntil(capture.flush()) on Vercel, await capture.flush() on AWS Lambda.

Every option is in the TypeScript and Python references.

Prepare streamed input

TypeScript SDK 0.14.0+ supports workspaceTools(ws, { prewake: true }); Python 0.10.0+ supports workspace_tools(ws, prewake=True). Call a VM tool's onInputStart() or on_input_start when the model starts its input, ahead of execution. These callbacks return immediately. Offline file reads and file-first workspaces have no callback. Default tools add no early hook.

Turn lifecycle in one call

With TypeScript SDK 0.16.0+, wrap a persistent workspace's agent turn in turn():

const ref = cloud.workspace('customer-42/main', { template: 'default' });
const answer = await ref.turn(async (workspace) => {
  const plan = await think(); // the hint overlaps model thinking
  return workspace.exec(plan.command);
}); // requests suspendWhenIdle({ afterSeconds: 0 }), including when the body throws

Python 0.12.0+ offers with ref.turn(): ... and async with ref.turn(): ..., also on an opened workspace. Async contexts offload lifecycle requests; exec and file methods remain synchronous.

0 frees RAM about 2 seconds after the turn; the next turn resumes, hidden while the model thinks. 60 keeps the workspace warm for a minute and bills that minute. In a measured agent integration, turn end to suspended took 1.9–2.4 seconds with 0, versus about 33 seconds with 30. With about 1.5 seconds of model thinking after the next turn's hint, its first command ran in 247–433 ms. Model thinking and running commands determine how much of the resume is hidden.

Test your application loop with the SDK fakes and the same assertions against Shardflux: Test your integration.

Isolate heavy work in cloud conversations

Native coding-agent conversations on coding 165 or newer have two additional built-in conversation tools. A thread can call conversation_isolate with a continuation describing its next step, commit its current edits, and end its turn. Its conversation identity and saved session move to a fresh VM; the continuation is delivered there once. Use conversation_spawn when the work should begin as a separate child. Separate VMs have independent local services and filesystems; shared project volumes and external databases still need independent test data. Existing resource quotas apply, and a separate VM does not imply a dedicated physical host.

After reviewing a finished worker, its parent or a related integration thread calls conversation_consolidate with the worker's id. Run it from the integrating conversation's clean branch. The tool merges the worker's exact saved commit, saves the integrated branch and transcript, then requests suspension of the worker's VM when idle. Run the project's integration tests on the result. The tool does not push to an external Git remote. Suspension respects the VM's existing agent idle grace, keepalives and active-work checks; it need not happen immediately when the merge returns.

A merge conflict remains in the integrating worktree for repair. Commit the resolution and retry the same call. Unfinished work, missing saved copies, pending input, active commands and VMs shared by other threads prevent the handoff. The worker's chat and saved work remain available; its VM follows the normal 48-hour retention period.

These tools also run inside a cloud conversation's shell:

sf-agent conversation isolate --args '{"continuation":"Run the integration suite with a local test database."}'
# End the current turn; the same session continues on its own VM.

# In the parent or integration thread, after the worker finishes:
sf-agent conversation consolidate --args '{"id":"conv_..."}'