Mercury CLI SDK

mercury-cli-sdk 0.1.0 is a zero-runtime-dependency client for Mercury CLI in TypeScript (requiring Node.js 22 or later) and Python (requiring Python 3.11 or later). The TypeScript and Python SDKs in sdk/ and sdk-py/ (mercury-cli-sdk on npm and PyPI) are licensed under the Apache License 2.0 (Apache-2.0) by their own sdk/LICENSE and sdk-py/LICENSE files and versioned separately from the CLI, and the Mercury CLI is licensed under the same Apache License 2.0 in LICENSE.md and NOTICE:

The SDK provides two doors into Mercury CLI:

run() (and run_sync() in Python) for one-shot turns with streamed rows (text, text_delta with partial, tool_call, tool_result, outcome), async iteration or typed listeners, structured output (schema), inline tools (tools), session resume (resume, continue in TypeScript and continue_ in Python), and cancellation (signal in TypeScript, cancel or interrupt() in Python).
connect() (and connect_sync() in Python) for a hosted multi-turn session with permission asks and elicitations answered by your code (onPermissionRequest / on_permission_request, onElicitation / on_elicitation), inline tools and custom agents (defineTool / define_tool, defineAgent / define_agent), session-wide and per-turn structured output (schema), file and conversation checkpoints (rewind), warm workers (prewarm, claim), session resume, and turn or process control (interrupt, terminate, kill, close, plus cooperative cancellation via { signal } / ToolContext.cancelled in tool callbacks).

Install

Install the TypeScript SDK from npm or the Python SDK from PyPI:

npm install mercury-cli-sdk
pip install mercury-cli-sdk

In Python, you can also run python -m pip install mercury-cli-sdk (on Windows, py -3 -m pip install mercury-cli-sdk).

What you need

Install Mercury CLI separately (Install and update) and sign in to a provider through the CLI (/logins or mercury auth login); the SDK uses Mercury's configuration and credentials.

The SDK finds your Mercury CLI installation automatically: MERCURY_BIN overrides discovery; otherwise it checks the managed install (current.txt under MERCURY_VERSIONS_DIR or versions in MERCURY_CONFIG_DIR, MERCURY_HOME or ~/.mercury; %USERPROFILE%\.mercury on Windows), mercury on PATH, the stable installer command (~/.local/bin/mercury; %LOCALAPPDATA%\Mercury\bin\mercury.cmd on Windows) and ~/.mercury/bin/mercury, and the Homebrew command on macOS and Linux. Both doors also accept an explicit executable or command prefix array (mercury).

Quickstart

TypeScript

import { run } from 'mercury-cli-sdk'

const task = run({ prompt: 'Explain the README', partial: true })
task.on('text_delta', row => process.stdout.write(row.text))
const result = await task.outcome
console.log(result.status, result.answer)

run() returns an async row iterator with typed listeners. Its session, outcome and exit are promises; rows retains received rows. Whole blocks arrive as text; partial: true also streams text_delta. outcome settles after the process exits and owned tool servers close.

Python

from mercury_cli_sdk import run_sync

print(run_sync('Explain the README').outcome.get('answer'))

For asynchronous programs and streaming, use run:

import asyncio
from mercury_cli_sdk import run

async def main():
    task = await run('Explain the README', partial=True)
    async for row in task:
        if row['type'] == 'text_delta':
            print(row['text'], end='', flush=True)
    result = await task.outcome
    print(result['status'], result.get('answer'))
asyncio.run(main())

run_sync returns RunResult(outcome, rows, exit). connect_sync returns a typed SyncSession with the same control methods and blocking row iteration.

Model selection

model uses the CLI's model ids and credentials. For a named OpenAI-compatible endpoint, pass compat/<endpoint>/<model> to run, run_sync or connect (for example run({ prompt: 'Explain the README', model: 'compat/my-server/model-id' }) in TypeScript or run_sync('Explain the README', model='compat/my-server/model-id') in Python), defineAgent / define_agent, and session.setModel / session.set_model (/llmhost registers a server as a named compat endpoint).

With Mercury 1.0.0-beta.42 or newer, compat, local and huggingface also resolve through the CLI's resolver for that family (mercury run --model local|compat|huggingface uses the same resolver). An unavailable class or an unknown named compat endpoint or model is refused instead of being replaced by another provider. A resumed session retains its recorded model (compat/<endpoint>/<model> included) across cold and warm session resume unless you explicitly supply another; an unavailable recorded model refuses the resume. Tool results, streamed rows and structured output use the same SDK interfaces on compat and Anthropic models, and reasoning and usage rows report what the engine supplies.

Host a session

TypeScript

import { connect } from 'mercury-cli-sdk'

const session = await connect({
  onPermissionRequest: () => ({ outcome: 'deny', message: 'Read-only host' }),
})
try {
  console.log((await session.prompt('Inspect the tests')).status)
  console.log((await session.prompt('Explain the result')).answer)
  console.log((await session.facts()).usage.total_cost_usd)
} finally { await session.close() }

Python

import asyncio
from mercury_cli_sdk import connect

async def main():
    async with await connect(on_permission_request=lambda ask, context: {'outcome': 'deny', 'message': 'Read-only host'}) as session:
        print((await session.prompt('Inspect the tests'))['status'])
        print((await session.prompt('Explain the result')).get('answer'))
        print((await session.facts())['usage']['total_cost_usd'])
asyncio.run(main())

prompt awaits its own correlated turn. steer queues priority now and followUp / follow_up queues later; withdraw(id) removes queued input, and sendNow() / send_now() sends input queued behind the running turn into that turn at once, as ctrl+s s does in the chat (Sessions). interrupt() ends the active turn, and setModel / set_model, setEffort / set_effort, setMode / set_mode and setKit / set_kit update the session. onReadTextFile / on_read_text_file and onWriteTextFile / on_write_text_file route native file tools to host buffers.

connect({ timeoutMs }) in TypeScript and connect(timeout=seconds) in Python bound initialization, and prompt(content, { timeoutMs }) / prompt(content, timeout=seconds) bound the whole prompt wait including queue acceptance (interruptOnAbort: true / interrupt_on_cancel=True interrupts the turn when the wait aborts, including on late acceptance), raising MercuryTimeoutError with method and the timeout; check facts() before resending work that may already have been accepted. assertCompleted(outcome) / assert_completed(outcome) raises MercuryOutcomeError (retaining outcome, errorClass / error_class and detail, with its message naming the tool and the runner's recovery instructions for granting permission or attaching a host) on non-completed outcomes while refusals raise MercuryRefusal. A closed stream while the runner is alive, or a call made during or after close() (including on SyncSession), fails immediately with MercuryConnectionError and settles all pending requests and row iterators, while malformed known notifications raise MercuryWireError and unknown future notifications remain available through on('notification', listener).

Inline tools, structured output and checkpoints

Inline tools

import { defineTool, run } from 'mercury-cli-sdk'

const greeting = defineTool({
  name: 'greeting', description: 'Return a greeting',
  inputSchema: { type: 'object', properties: {} },
  execute: (_input, { signal }) => signal.aborted ? 'Cancelled' : 'Hello',
})
const result = await run({ prompt: 'Use greeting to say hello', tools: [greeting] }).outcome
console.log(result.status, result.answer)

import asyncio
from mercury_cli_sdk import define_tool, run

async def main():
    greeting = define_tool('greeting', 'Return a greeting', {}, lambda _: 'Hello')
    task = await run('Use greeting to say hello', tools=[greeting])
    result = await task.outcome
    print(result['status'], result.get('answer'))
asyncio.run(main())

Both doors own an authenticated loopback MCP server for tools, close it on exit or failed startup, merge it with mcp, and add mcp__sdk-tools__<name> to allowedTools / allowed_tools.

Structured output, file checkpoints and version requirements

A schema accepts JSON Schema, JSON text, a schema file path or structuredSchema(jsonSchema, decode) / structured_schema(json_schema, decode), returning the decoded object in outcome.structured (and rejecting invalid schemas before any model turn is spent with MercurySchemaError). On run it passes --schema; on connect it travels on initialize and --schema; on session.prompt(content, { schema }) / session.prompt(content, schema=...) it sets that turn's schema alone.

connect asks the CLI to capture file checkpoints (file_checkpoints: true on initialize; checkpoints: false / checkpoints=False opts out). After any accepted prompt input, lastPromptId / last_prompt_id is its user-message id, facts().file_checkpoints lists restorable user-message ids along with tracked files and any undo target, rewind can preview (dry_run) and restore code, conversation or both, and session/file_restore restores a single file (action: "restore") or undoes the last file restore (action: "undo").

Per-turn structured output and file checkpoints over initialize need Mercury 1.0.0-beta.41 or newer. On 1.0.0-beta.40, an older runner ignores a per-turn schema field, so the SDK refuses it before sending with MercuryError and says to run mercury update (session.supports.perTurnSchema / session.supports['per_turn_schema'] beside session.runnerVersion / session.runner_version reports support in advance, and FEATURE_FLOORS lists each floor). For file checkpoints on an older runner, initialize answers no file_checkpoints, so session.fileCheckpoints / session.file_checkpoints is undefined / None, session.supports.fileCheckpoints / session.supports['file_checkpoints'] is false / False, and facts().file_checkpoints reads capture: false with no source.

Test without a model

startFixtureServer (mercury-cli-sdk/testing) and start_fixture_server (mercury_cli_sdk.testing) replay scripted rows and JSON-RPC replies locally at zero cost: fixtures use neither a CLI nor model credentials and do not simulate the CLI's internal behaviour.

import { run } from 'mercury-cli-sdk'
import { startFixtureServer, fixtureSessionRow, fixtureOutcomeRow } from 'mercury-cli-sdk/testing'

const fixture = await startFixtureServer({ rows: [fixtureSessionRow(), fixtureOutcomeRow({ answer: 'ok' })] })
try {
  const result = await run({ mercury: fixture.mercury, prompt: 'test' }).outcome
  if (result.answer !== 'ok') throw new Error('unexpected answer')
} finally { await fixture.close() }

import asyncio
from mercury_cli_sdk import run
from mercury_cli_sdk.testing import start_fixture_server, fixture_session_row, fixture_outcome_row

async def main():
    rows = [fixture_session_row(), fixture_outcome_row(answer='ok')]
    async with await start_fixture_server({'rows': rows}) as fixture:
        task = await run('test', mercury=fixture.mercury)
        assert (await task.outcome)['answer'] == 'ok'
asyncio.run(main())

Automation · Command-line reference · Sessions · Install and update