claude-code-api

Python wrapper around the claude CLI for subscription-mode (no API key) backends. Drives one long-running claude per conversation and yields typed events; the public surface is Anthropic-Messages-API shaped so a gateway in front of it is a one-liner serializer away.

Two transports, picked with BackendOptions.transport:

• stream_json — claude -p with stream-json on stdin and stdout. Prompts go into a pipe, events come back as they happen, and a native result record closes each turn. Prefer this.
• pty (default, for backwards compatibility) — drives the interactive TUI through a pseudo-tty and tails the session JSONL. It has to guess when the TUI is ready to accept a paste, re-press Enter when the paste is swallowed, and poll a file for output. Keep it only if you need something the TUI alone has (e.g. --remote-control).

Measured on the same host, same model, same prompt — cold turn, single reply: pty spends 2.3s on TUI readiness before the prompt is even submitted (first event 2.5s, turn 5.2s); stream_json spends none (first event 0.3s, first token 1.9s, turn 2.7s). The gap widens on slow hardware, which is what the PTY transport's 60s startup_delay cap exists to survive.

Not affiliated with Anthropic. You need a working subscription, the claude CLI on PATH, and to have run claude /login once.

Install

As a library inside another project:

uv add "claude-code-api @ git+https://git.kotikot.com/beaver/claude-code-api"

The runtime needs only ptyprocess.

Use

import asyncio
from claude_code_api import BackendOptions, ClaudeCodeBackend

async def main() -> None:
    opts = BackendOptions(
        cwd="/path/to/project",
        dangerously_skip_permissions=True,
        transport="stream_json",
        include_partial_messages=True,  # token-level deltas, stream_json only
    )
    async with ClaudeCodeBackend(opts) as backend:
        async for event in backend.complete(
            [{"role": "user", "content": "say hi"}]
        ):
            print(event)

asyncio.run(main())

Multi-turn works by construction — append the assistant reply + a fresh user message to the same messages list and call complete() again. The backend fingerprints messages[:-1], finds the live PTY from the previous turn, and reuses it (so the server-side prompt cache stays warm):

history = [{"role": "user", "content": "remember Beaver"}]
async for ev in backend.complete(history): ...

history += [
    {"role": "assistant", "content": [{"type": "text", "text": "OK"}]},
    {"role": "user", "content": "what was the codeword?"},
]
async for ev in backend.complete(history): ...

Public surface

Events (Anthropic-style, vendored to keep the dep tree empty): AssistantMessage, UserMessage, SystemMessage, ResultMessage, TextBlock, ThinkingBlock, ToolUseBlock, ToolResultBlock.

Errors: BackendError (root), AuthError, ProcessError, CLINotFoundError, RateLimitError, SessionError, MessageParseError.

Backend: ClaudeCodeBackend(opts).complete(messages) is an async generator of events. BackendOptions exposes model / system prompt / allowed-tools / mcp_servers / permission mode / history injection mode.

Lower layers (PtyClaudeProcess, JsonlWatcher, TurnManager, normalize) are re-exported for callers that want to assemble their own session orchestration.

How a turn works

Steps 1–3 are shared by both transports:

1. The backend looks up a live session by hash_history(messages[:-1]). If one matches, the new user message goes straight to it.
2. If nothing matches and messages[:-1] is empty, a fresh claude is spawned with a brand-new --session-id.
3. If messages[:-1] is non-empty (a continuation we don't have a live process for — e.g. after restart), the backend writes a hand-crafted JSONL transcript at ~/.claude/projects/<key>/<id>.jsonl and spawns claude --resume <id>. That is the native_jsonl injection mode; the fallback is concat_message, which folds the prior history into one large first prompt.

Then, on stream_json:

4. The prompt is written to stdin as one JSON line. No readiness wait — the CLI buffers stdin, so this works even before it has finished booting.
5. stdout is read line by line and each record normalized into a typed Event. With include_partial_messages the raw Anthropic message_stream events also come through as StreamEvent, with index rebased across the turn's several API requests.
6. The turn closes on the native result record, which supplies the ResultMessage verbatim — aggregate usage, real durations, cost.

Or, on pty:

4. The PTY's stdout is drained continuously by a background thread; we never read events from there. The prompt is pasted in bracketed-paste framing followed by a separate Enter, and re-sent if the JSONL shows no sign it registered. The JSONL file is tailed at 100ms cadence and each new record is normalized into a typed Event.
5. The turn closes on the first assistant record with stop_reason ∈ {end_turn, max_tokens, stop_sequence, refusal}. A ResultMessage is synthesized from its usage and yielded last.

Examples

• examples/basic_usage.py — one turn, real claude.
• examples/multi_turn.py — two turns sharing one live PTY.
• examples/mcp_tool.py — wire up the bundled echo MCP server and let the model call it.

Tests

uv run pytest                  # unit tests (fast, no real claude)
RUN_CLAUDE_SMOKE=1 uv run pytest tests/test_pty.py tests/test_turn.py tests/test_backend.py tests/test_stream.py

The smoke-marked tests spawn a real claude process and need a logged-in subscription on the host.