mirrors/claude-code-sdk-python
1
0
Fork 0
mirror of https://github.com/anthropics/claude-code-sdk-python.git synced 2025-07-07 14:45:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Initial Python SDK import

Browse source
This commit is contained in:
Lina Tawfik 2025-06-12 00:16:19 -07:00
parent 19c71ae9ca
commit 6ca3514261
No known key found for this signature in database
22 changed files with 1774 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

33
.github/workflows/lint.yml vendored Normal file
View file

@ -0,0 +1,33 @@
name: Lint
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]"
- name: Run ruff
run: |
ruff check src/ tests/
ruff format --check src/ tests/
- name: Run mypy
run: |
mypy src/

111
.github/workflows/publish.yml vendored Normal file
View file

@ -0,0 +1,111 @@
name: Publish to PyPI
on:
workflow_dispatch:
inputs:
version:
description: 'Version to publish (e.g., 0.1.0)'
required: true
type: string
test_pypi:
description: 'Publish to Test PyPI first'
required: false
type: boolean
default: true
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.10', '3.11', '3.12', '3.13']
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]"
- name: Run tests
run: |
python -m pytest tests/ -v
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]"
- name: Run ruff
run: |
ruff check src/ tests/
ruff format --check src/ tests/
- name: Run mypy
run: |
mypy src/
publish:
needs: [test, lint]
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Update version
run: |
# Update version in pyproject.toml
sed -i 's/version = ".*"/version = "${{ github.event.inputs.version }}"/' pyproject.toml
sed -i 's/__version__ = ".*"/__version__ = "${{ github.event.inputs.version }}"/' src/claude_code_sdk/__init__.py
- name: Install build dependencies
run: |
python -m pip install --upgrade pip
pip install build twine
- name: Build package
run: python -m build
- name: Check package
run: twine check dist/*
- name: Publish to Test PyPI
if: ${{ github.event.inputs.test_pypi == 'true' }}
env:
TWINE_USERNAME: __token__
TWINE_PASSWORD: ${{ secrets.TEST_PYPI_API_TOKEN }}
run: |
twine upload --repository testpypi dist/*
echo "Package published to Test PyPI"
echo "Install with: pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple/ claude-code-sdk==${{ github.event.inputs.version }}"
- name: Publish to PyPI
env:
TWINE_USERNAME: __token__
TWINE_PASSWORD: ${{ secrets.PYPI_API_TOKEN }}
run: |
twine upload dist/*
echo "Package published to PyPI"
echo "Install with: pip install claude-code-sdk==${{ github.event.inputs.version }}"

37
.github/workflows/test.yml vendored Normal file
View file

@ -0,0 +1,37 @@
name: Test
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.10', '3.11', '3.12', '3.13']
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[dev]"
- name: Run tests
run: |
python -m pytest tests/ -v --cov=claude_code_sdk --cov-report=xml
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v4
with:
file: ./coverage.xml
fail_ci_if_error: true

49
.gitignore vendored Normal file
View file

@ -0,0 +1,49 @@
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# Virtual environments
venv/
ENV/
env/
.venv
# IDEs
.vscode/
.idea/
*.swp
*.swo
*~
# Testing
.tox/
.coverage
.coverage.*
.cache
.pytest_cache/
htmlcov/
# Type checking
.mypy_cache/
.dmypy.json
dmypy.json
.pyre/

21
LICENSE Normal file
View file

@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Anthropic, PBC
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

133
README.md
View file

@ -1 +1,132 @@
# claude-code-sdk-python
# Claude Code SDK for Python
Python SDK for Claude Code. See the [Claude Code SDK documentation](https://docs.anthropic.com/en/docs/claude-code/sdk) for more information.
## Installation
```bash
pip install claude-code-sdk
```
**Prerequisites:**
- Python 3.10+
- Node.js
- Claude Code: `npm install -g @anthropic-ai/claude-code`
## Quick Start
```python
import anyio
from claude_code_sdk import query
async def main():
async for message in query(prompt="What is 2 + 2?"):
print(message)
anyio.run(main)
```
## Usage
### Basic Query
```python
from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, TextBlock
# Simple query
async for message in query(prompt="Hello Claude"):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
# With options
options = ClaudeCodeOptions(
system_prompt="You are a helpful assistant",
max_turns=1
)
async for message in query(prompt="Tell me a joke", options=options):
print(message)
```
### Using Tools
```python
options = ClaudeCodeOptions(
allowed_tools=["Read", "Write", "Bash"],
permission_mode='acceptEdits' # auto-accept file edits
)
async for message in query(
prompt="Create a hello.py file",
options=options
):
# Process tool use and results
pass
```
### Working Directory
```python
from pathlib import Path
options = ClaudeCodeOptions(
cwd="/path/to/project" # or Path("/path/to/project")
)
```
## API Reference
### `query(prompt, options=None)`
Main async function for querying Claude.
**Parameters:**
- `prompt` (str): The prompt to send to Claude
- `options` (ClaudeCodeOptions): Optional configuration
**Returns:** AsyncIterator[Message] - Stream of response messages
### Types
See [src/claude_code_sdk/types.py](src/claude_code_sdk/types.py) for complete type definitions:
- `ClaudeCodeOptions` - Configuration options
- `AssistantMessage`, `UserMessage`, `SystemMessage`, `ResultMessage` - Message types
- `TextBlock`, `ToolUseBlock`, `ToolResultBlock` - Content blocks
## Error Handling
```python
from claude_code_sdk import (
ClaudeSDKError, # Base error
CLINotFoundError, # Claude Code not installed
CLIConnectionError, # Connection issues
ProcessError, # Process failed
CLIJSONDecodeError, # JSON parsing issues
)
try:
async for message in query(prompt="Hello"):
pass
except CLINotFoundError:
print("Please install Claude Code")
except ProcessError as e:
print(f"Process failed with exit code: {e.exit_code}")
except CLIJSONDecodeError as e:
print(f"Failed to parse response: {e}")
```
See [src/claude_code_sdk/_errors.py](src/claude_code_sdk/_errors.py) for all error types.
## Available Tools
See the [Claude Code documentation](https://docs.anthropic.com/en/docs/claude-code/security#tools-available-to-claude) for a complete list of available tools.
## Examples
See [examples/quick_start.py](examples/quick_start.py) for a complete working example.
## License
MIT

77
examples/quick_start.py Normal file
View file

@ -0,0 +1,77 @@
#!/usr/bin/env python3
"""Quick start example for Claude Code SDK."""
import anyio
from claude_code_sdk import (
AssistantMessage,
ClaudeCodeOptions,
ResultMessage,
TextBlock,
query,
)
async def basic_example():
"""Basic example - simple question."""
print("=== Basic Example ===")
async for message in query(prompt="What is 2 + 2?"):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
print()
async def with_options_example():
"""Example with custom options."""
print("=== With Options Example ===")
options = ClaudeCodeOptions(
system_prompt="You are a helpful assistant that explains things simply.",
max_turns=1,
)
async for message in query(
prompt="Explain what Python is in one sentence.",
options=options
):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
print()
async def with_tools_example():
"""Example using tools."""
print("=== With Tools Example ===")
options = ClaudeCodeOptions(
allowed_tools=["Read", "Write"],
system_prompt="You are a helpful file assistant.",
)
async for message in query(
prompt="Create a file called hello.txt with 'Hello, World!' in it",
options=options
):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
elif isinstance(message, ResultMessage) and message.cost_usd > 0:
print(f"\nCost: ${message.cost_usd:.4f}")
print()
async def main():
"""Run all examples."""
await basic_example()
await with_options_example()
await with_tools_example()
if __name__ == "__main__":
anyio.run(main)

106
pyproject.toml Normal file
View file

@ -0,0 +1,106 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "claude-code-sdk"
version = "0.0.10"
description = "Python SDK for Claude Code"
readme = "README.md"
requires-python = ">=3.10"
license = {text = "MIT"}
authors = [
{name = "Anthropic", email = "support@anthropic.com"},
]
classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Typing :: Typed",
]
keywords = ["claude", "ai", "sdk", "anthropic"]
dependencies = [
"anyio>=4.0.0",
"typing_extensions>=4.0.0; python_version<'3.11'",
]
[project.optional-dependencies]
dev = [
"pytest>=7.0.0",
"pytest-asyncio>=0.20.0",
"anyio[trio]>=4.0.0",
"pytest-cov>=4.0.0",
"mypy>=1.0.0",
"ruff>=0.1.0",
]
[project.urls]
Homepage = "https://github.com/anthropics/claude-code-sdk-python"
Documentation = "https://docs.anthropic.com/en/docs/claude-code/sdk"
Issues = "https://github.com/anthropics/claude-code-sdk-python/issues"
[tool.hatch.build.targets.wheel]
packages = ["src/claude_code_sdk"]
[tool.hatch.build.targets.sdist]
include = [
"/src",
"/tests",
"/README.md",
"/LICENSE",
]
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["src"]
addopts = [
"--import-mode=importlib",
]
[tool.pytest-asyncio]
asyncio_mode = "auto"
[tool.mypy]
python_version = "3.10"
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
disallow_untyped_decorators = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_no_return = true
warn_unreachable = true
strict_equality = true
[tool.ruff]
target-version = "py310"
line-length = 88
[tool.ruff.lint]
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # pyflakes
"I", # isort
"N", # pep8-naming
"UP", # pyupgrade
"B", # flake8-bugbear
"C4", # flake8-comprehensions
"PTH", # flake8-use-pathlib
"SIM", # flake8-simplify
]
ignore = [
"E501", # line too long (handled by formatter)
]
[tool.ruff.lint.isort]
known-first-party = ["claude_code_sdk"]

102
src/claude_code_sdk/__init__.py Normal file
View file

@ -0,0 +1,102 @@
"""Claude SDK for Python."""
import os
from collections.abc import AsyncIterator
from ._errors import (
ClaudeSDKError,
CLIConnectionError,
CLIJSONDecodeError,
CLINotFoundError,
ProcessError,
)
from ._internal.client import InternalClient
from .types import (
AssistantMessage,
ClaudeCodeOptions,
ContentBlock,
McpServerConfig,
Message,
PermissionMode,
ResultMessage,
SystemMessage,
TextBlock,
ToolResultBlock,
ToolUseBlock,
UserMessage,
)
__version__ = "0.0.10"
__all__ = [
# Main function
"query",
# Types
"PermissionMode",
"McpServerConfig",
"UserMessage",
"AssistantMessage",
"SystemMessage",
"ResultMessage",
"Message",
"ClaudeCodeOptions",
"TextBlock",
"ToolUseBlock",
"ToolResultBlock",
"ContentBlock",
# Errors
"ClaudeSDKError",
"CLIConnectionError",
"CLINotFoundError",
"ProcessError",
"CLIJSONDecodeError",
]
async def query(
*, prompt: str, options: ClaudeCodeOptions | None = None
) -> AsyncIterator[Message]:
"""
Query Claude Code.
Python SDK for interacting with Claude Code.
Args:
prompt: The prompt to send to Claude
options: Optional configuration (defaults to ClaudeCodeOptions() if None).
Set options.permission_mode to control tool execution:
- 'default': CLI prompts for dangerous tools
- 'acceptEdits': Auto-accept file edits
- 'bypassPermissions': Allow all tools (use with caution)
Set options.cwd for working directory.
Yields:
Messages from the conversation
Example:
```python
# Simple usage
async for message in query(prompt="Hello"):
print(message)
# With options
async for message in query(
prompt="Hello",
options=ClaudeCodeOptions(
system_prompt="You are helpful",
cwd="/home/user"
)
):
print(message)
```
"""
if options is None:
options = ClaudeCodeOptions()
os.environ["CLAUDE_CODE_ENTRYPOINT"] = "sdk-py"
client = InternalClient()
async for message in client.process_query(prompt=prompt, options=options):
yield message

46
src/claude_code_sdk/_errors.py Normal file
View file

@ -0,0 +1,46 @@
"""Error types for Claude SDK."""
class ClaudeSDKError(Exception):
"""Base exception for all Claude SDK errors."""
class CLIConnectionError(ClaudeSDKError):
"""Raised when unable to connect to Claude Code."""
class CLINotFoundError(CLIConnectionError):
"""Raised when Claude Code is not found or not installed."""
def __init__(
self, message: str = "Claude Code not found", cli_path: str | None = None
):
if cli_path:
message = f"{message}: {cli_path}"
super().__init__(message)
class ProcessError(ClaudeSDKError):
"""Raised when the CLI process fails."""
def __init__(
self, message: str, exit_code: int | None = None, stderr: str | None = None
):
self.exit_code = exit_code
self.stderr = stderr
if exit_code is not None:
message = f"{message} (exit code: {exit_code})"
if stderr:
message = f"{message}\nError output: {stderr}"
super().__init__(message)
class CLIJSONDecodeError(ClaudeSDKError):
"""Raised when unable to decode JSON from CLI output."""
def __init__(self, line: str, original_error: Exception):
self.line = line
self.original_error = original_error
super().__init__(f"Failed to decode JSON: {line[:100]}...")

1
src/claude_code_sdk/_internal/__init__.py Normal file
View file

@ -0,0 +1 @@
"""Internal implementation details."""

99
src/claude_code_sdk/_internal/client.py Normal file
View file

@ -0,0 +1,99 @@
"""Internal client implementation."""
from collections.abc import AsyncIterator
from typing import Any
from ..types import (
AssistantMessage,
ClaudeCodeOptions,
ContentBlock,
Message,
ResultMessage,
SystemMessage,
TextBlock,
ToolResultBlock,
ToolUseBlock,
UserMessage,
)
from .transport.subprocess_cli import SubprocessCLITransport
class InternalClient:
"""Internal client implementation."""
def __init__(self) -> None:
"""Initialize the internal client."""
async def process_query(
self, prompt: str, options: ClaudeCodeOptions
) -> AsyncIterator[Message]:
"""Process a query through transport."""
transport = SubprocessCLITransport(prompt=prompt, options=options)
try:
await transport.connect()
async for data in transport.receive_messages():
message = self._parse_message(data)
if message:
yield message
finally:
await transport.disconnect()
def _parse_message(self, data: dict[str, Any]) -> Message | None:
"""Parse message from CLI output, trusting the structure."""
match data["type"]:
case "user":
# Extract just the content from the nested structure
return UserMessage(
content=data["message"]["content"]
)
case "assistant":
# Parse content blocks
content_blocks: list[ContentBlock] = []
for block in data["message"]["content"]:
match block["type"]:
case "text":
content_blocks.append(TextBlock(text=block["text"]))
case "tool_use":
content_blocks.append(ToolUseBlock(
id=block["id"],
name=block["name"],
input=block["input"]
))
case "tool_result":
content_blocks.append(ToolResultBlock(
tool_use_id=block["tool_use_id"],
content=block.get("content"),
is_error=block.get("is_error")
))
return AssistantMessage(content=content_blocks)
case "system":
return SystemMessage(
subtype=data["subtype"],
data=data # Pass through all data
)
case "result":
# Map total_cost to total_cost_usd for consistency
return ResultMessage(
subtype=data["subtype"],
cost_usd=data["cost_usd"],
duration_ms=data["duration_ms"],
duration_api_ms=data["duration_api_ms"],
is_error=data["is_error"],
num_turns=data["num_turns"],
session_id=data["session_id"],
total_cost_usd=data["total_cost"],
usage=data.get("usage"),
result=data.get("result")
)
case _:
return None

39
src/claude_code_sdk/_internal/transport/__init__.py Normal file
View file

@ -0,0 +1,39 @@
"""Transport implementations for Claude SDK."""
from abc import ABC, abstractmethod
from collections.abc import AsyncIterator
from typing import Any
class Transport(ABC):
"""Abstract transport for Claude communication."""
@abstractmethod
async def connect(self) -> None:
"""Initialize connection."""
pass
@abstractmethod
async def disconnect(self) -> None:
"""Close connection."""
pass
@abstractmethod
async def send_request(
self, messages: list[dict[str, Any]], options: dict[str, Any]
) -> None:
"""Send request to Claude."""
pass
@abstractmethod
def receive_messages(self) -> AsyncIterator[dict[str, Any]]:
"""Receive messages from Claude."""
pass
@abstractmethod
def is_connected(self) -> bool:
"""Check if transport is connected."""
pass
__all__ = ["Transport"]

213
src/claude_code_sdk/_internal/transport/subprocess_cli.py Normal file
View file

@ -0,0 +1,213 @@
"""Subprocess transport implementation using Claude Code CLI."""
import json
import os
import shutil
from collections.abc import AsyncIterator
from pathlib import Path
from subprocess import PIPE
from typing import Any
import anyio
from anyio.abc import Process
from anyio.streams.text import TextReceiveStream
from ..._errors import CLIConnectionError, CLINotFoundError, ProcessError
from ..._errors import CLIJSONDecodeError as SDKJSONDecodeError
from ...types import ClaudeCodeOptions
from . import Transport
class SubprocessCLITransport(Transport):
"""Subprocess transport using Claude Code CLI."""
def __init__(
self, prompt: str, options: ClaudeCodeOptions, cli_path: str | Path | None = None
):
self._prompt = prompt
self._options = options
self._cli_path = str(cli_path) if cli_path else self._find_cli()
self._cwd = str(options.cwd) if options.cwd else None
self._process: Process | None = None
self._stdout_stream: TextReceiveStream | None = None
self._stderr_stream: TextReceiveStream | None = None
def _find_cli(self) -> str:
"""Find Claude Code CLI binary."""
if cli := shutil.which("claude"):
return cli
locations = [
Path.home() / ".npm-global/bin/claude",
Path("/usr/local/bin/claude"),
Path.home() / ".local/bin/claude",
Path.home() / "node_modules/.bin/claude",
Path.home() / ".yarn/bin/claude",
]
for path in locations:
if path.exists() and path.is_file():
return str(path)
node_installed = shutil.which("node") is not None
if not node_installed:
error_msg = "Claude Code requires Node.js, which is not installed.\n\n"
error_msg += "Install Node.js from: https://nodejs.org/\n"
error_msg += "\nAfter installing Node.js, install Claude Code:\n"
error_msg += " npm install -g @anthropic-ai/claude-code"
raise CLINotFoundError(error_msg)
raise CLINotFoundError(
"Claude Code not found. Install with:\n"
" npm install -g @anthropic-ai/claude-code\n"
"\nIf already installed locally, try:\n"
' export PATH="$HOME/node_modules/.bin:$PATH"\n'
"\nOr specify the path when creating transport:\n"
" SubprocessCLITransport(..., cli_path='/path/to/claude')"
)
def _build_command(self) -> list[str]:
"""Build CLI command with arguments."""
cmd = [self._cli_path, "--output-format", "stream-json", "--verbose"]
if self._options.system_prompt:
cmd.extend(["--system-prompt", self._options.system_prompt])
if self._options.append_system_prompt:
cmd.extend(["--append-system-prompt", self._options.append_system_prompt])
if self._options.allowed_tools:
cmd.extend(["--allowedTools", ",".join(self._options.allowed_tools)])
if self._options.max_turns:
cmd.extend(["--max-turns", str(self._options.max_turns)])
if self._options.disallowed_tools:
cmd.extend(["--disallowedTools", ",".join(self._options.disallowed_tools)])
if self._options.model:
cmd.extend(["--model", self._options.model])
if self._options.permission_prompt_tool_name:
cmd.extend(
["--permission-prompt-tool", self._options.permission_prompt_tool_name]
)
if self._options.permission_mode:
cmd.extend(["--permission-mode", self._options.permission_mode])
if self._options.continue_conversation:
cmd.append("--continue")
if self._options.resume:
cmd.extend(["--resume", self._options.resume])
if self._options.mcp_servers:
cmd.extend(
["--mcp-config", json.dumps({"mcpServers": self._options.mcp_servers})]
)
cmd.extend(["--print", self._prompt])
return cmd
async def connect(self) -> None:
"""Start subprocess."""
if self._process:
return
cmd = self._build_command()
try:
self._process = await anyio.open_process(
cmd,
stdin=None,
stdout=PIPE,
stderr=PIPE,
cwd=self._cwd,
env={**os.environ, "CLAUDE_CODE_ENTRYPOINT": "sdk-py"},
)
if self._process.stdout:
self._stdout_stream = TextReceiveStream(self._process.stdout)
if self._process.stderr:
self._stderr_stream = TextReceiveStream(self._process.stderr)
except FileNotFoundError as e:
raise CLINotFoundError(f"Claude Code not found at: {self._cli_path}") from e
except Exception as e:
raise CLIConnectionError(f"Failed to start Claude Code: {e}") from e
async def disconnect(self) -> None:
"""Terminate subprocess."""
if not self._process:
return
if self._process.returncode is None:
try:
self._process.terminate()
with anyio.fail_after(5.0):
await self._process.wait()
except TimeoutError:
self._process.kill()
await self._process.wait()
except ProcessLookupError:
pass
self._process = None
self._stdout_stream = None
self._stderr_stream = None
async def send_request(self, messages: list[Any], options: dict[str, Any]) -> None:
"""Not used for CLI transport - args passed via command line."""
async def receive_messages(self) -> AsyncIterator[dict[str, Any]]:
"""Receive messages from CLI."""
if not self._process or not self._stdout_stream:
raise CLIConnectionError("Not connected")
stderr_lines = []
async def read_stderr() -> None:
"""Read stderr in background."""
if self._stderr_stream:
try:
async for line in self._stderr_stream:
stderr_lines.append(line.strip())
except anyio.ClosedResourceError:
pass
async with anyio.create_task_group() as tg:
tg.start_soon(read_stderr)
try:
async for line in self._stdout_stream:
line_str = line.strip()
if not line_str:
continue
try:
data = json.loads(line_str)
yield data
except json.JSONDecodeError as e:
if line_str.startswith("{") or line_str.startswith("["):
raise SDKJSONDecodeError(line_str, e) from e
continue
except anyio.ClosedResourceError:
pass
finally:
tg.cancel_scope.cancel()
await self._process.wait()
if self._process.returncode is not None and self._process.returncode != 0:
stderr_output = "\n".join(stderr_lines)
if stderr_output and "error" in stderr_output.lower():
raise ProcessError(
"CLI process failed",
exit_code=self._process.returncode,
stderr=stderr_output,
)
def is_connected(self) -> bool:
"""Check if subprocess is running."""
return self._process is not None and self._process.returncode is None

0
src/claude_code_sdk/py.typed Normal file
View file

100
src/claude_code_sdk/types.py Normal file
View file

@ -0,0 +1,100 @@
"""Type definitions for Claude SDK."""
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Literal, TypedDict
from typing_extensions import NotRequired # For Python < 3.11 compatibility
# Permission modes
PermissionMode = Literal["default", "acceptEdits", "bypassPermissions"]
# MCP Server config
class McpServerConfig(TypedDict):
"""MCP server configuration."""
transport: list[str]
env: NotRequired[dict[str, Any]]
# Content block types
@dataclass
class TextBlock:
"""Text content block."""
text: str
@dataclass
class ToolUseBlock:
"""Tool use content block."""
id: str
name: str
input: dict[str, Any]
@dataclass
class ToolResultBlock:
"""Tool result content block."""
tool_use_id: str
content: str | list[dict[str, Any]] | None = None
is_error: bool | None = None
ContentBlock = TextBlock | ToolUseBlock | ToolResultBlock
# Message types
@dataclass
class UserMessage:
"""User message."""
content: str
@dataclass
class AssistantMessage:
"""Assistant message with content blocks."""
content: list[ContentBlock]
@dataclass
class SystemMessage:
"""System message with metadata."""
subtype: str
data: dict[str, Any]
@dataclass
class ResultMessage:
"""Result message with cost and usage information."""
subtype: str
cost_usd: float
duration_ms: int
duration_api_ms: int
is_error: bool
num_turns: int
session_id: str
total_cost_usd: float
usage: dict[str, Any] | None = None
result: str | None = None
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage
@dataclass
class ClaudeCodeOptions:
"""Query options for Claude SDK."""
allowed_tools: list[str] = field(default_factory=list)
max_thinking_tokens: int = 8000
system_prompt: str | None = None
append_system_prompt: str | None = None
mcp_tools: list[str] = field(default_factory=list)
mcp_servers: dict[str, McpServerConfig] = field(default_factory=dict)
permission_mode: PermissionMode | None = None
continue_conversation: bool = False
resume: str | None = None
max_turns: int | None = None
disallowed_tools: list[str] = field(default_factory=list)
model: str | None = None
permission_prompt_tool_name: str | None = None
cwd: str | Path | None = None

4
tests/conftest.py Normal file
View file

@ -0,0 +1,4 @@
"""Pytest configuration for tests."""
# No async plugin needed since we're using sync tests with anyio.run()

115
tests/test_client.py Normal file
View file

@ -0,0 +1,115 @@
"""Tests for Claude SDK client functionality."""
from unittest.mock import AsyncMock, patch
import anyio
from claude_code_sdk import AssistantMessage, ClaudeCodeOptions, query
from claude_code_sdk.types import TextBlock
class TestQueryFunction:
"""Test the main query function."""
def test_query_single_prompt(self):
"""Test query with a single prompt."""
async def _test():
with patch('claude_code_sdk._internal.client.InternalClient.process_query') as mock_process:
# Mock the async generator
async def mock_generator():
yield AssistantMessage(
content=[TextBlock(text="4")]
)
mock_process.return_value = mock_generator()
messages = []
async for msg in query(prompt="What is 2+2?"):
messages.append(msg)
assert len(messages) == 1
assert isinstance(messages[0], AssistantMessage)
assert messages[0].content[0].text == "4"
anyio.run(_test)
def test_query_with_options(self):
"""Test query with various options."""
async def _test():
with patch('claude_code_sdk._internal.client.InternalClient.process_query') as mock_process:
async def mock_generator():
yield AssistantMessage(
content=[TextBlock(text="Hello!")]
)
mock_process.return_value = mock_generator()
options = ClaudeCodeOptions(
allowed_tools=["Read", "Write"],
system_prompt="You are helpful",
permission_mode="acceptEdits",
max_turns=5
)
messages = []
async for msg in query(
prompt="Hi",
options=options
):
messages.append(msg)
# Verify process_query was called with correct prompt and options
mock_process.assert_called_once()
call_args = mock_process.call_args
assert call_args[1]['prompt'] == "Hi"
assert call_args[1]['options'] == options
anyio.run(_test)
def test_query_with_cwd(self):
"""Test query with custom working directory."""
async def _test():
with patch('claude_code_sdk._internal.client.SubprocessCLITransport') as mock_transport_class:
mock_transport = AsyncMock()
mock_transport_class.return_value = mock_transport
# Mock the message stream
async def mock_receive():
yield {
"type": "assistant",
"message": {
"role": "assistant",
"content": [{"type": "text", "text": "Done"}]
}
}
yield {
"type": "result",
"subtype": "success",
"cost_usd": 0.001,
"duration_ms": 1000,
"duration_api_ms": 800,
"is_error": False,
"num_turns": 1,
"session_id": "test-session",
"total_cost": 0.001
}
mock_transport.receive_messages = mock_receive
mock_transport.connect = AsyncMock()
mock_transport.disconnect = AsyncMock()
options = ClaudeCodeOptions(cwd="/custom/path")
messages = []
async for msg in query(
prompt="test",
options=options
):
messages.append(msg)
# Verify transport was created with correct parameters
mock_transport_class.assert_called_once()
call_kwargs = mock_transport_class.call_args.kwargs
assert call_kwargs['prompt'] == "test"
assert call_kwargs['options'].cwd == "/custom/path"
anyio.run(_test)

52
tests/test_errors.py Normal file
View file

@ -0,0 +1,52 @@
"""Tests for Claude SDK error handling."""
from claude_code_sdk import (
ClaudeSDKError,
CLIConnectionError,
CLIJSONDecodeError,
CLINotFoundError,
ProcessError,
)
class TestErrorTypes:
"""Test error types and their properties."""
def test_base_error(self):
"""Test base ClaudeSDKError."""
error = ClaudeSDKError("Something went wrong")
assert str(error) == "Something went wrong"
assert isinstance(error, Exception)
def test_cli_not_found_error(self):
"""Test CLINotFoundError."""
error = CLINotFoundError("Claude Code not found")
assert isinstance(error, ClaudeSDKError)
assert "Claude Code not found" in str(error)
def test_connection_error(self):
"""Test CLIConnectionError."""
error = CLIConnectionError("Failed to connect to CLI")
assert isinstance(error, ClaudeSDKError)
assert "Failed to connect to CLI" in str(error)
def test_process_error(self):
"""Test ProcessError with exit code and stderr."""
error = ProcessError("Process failed", exit_code=1, stderr="Command not found")
assert error.exit_code == 1
assert error.stderr == "Command not found"
assert "Process failed" in str(error)
assert "exit code: 1" in str(error)
assert "Command not found" in str(error)
def test_json_decode_error(self):
"""Test CLIJSONDecodeError."""
import json
try:
json.loads("{invalid json}")
except json.JSONDecodeError as e:
error = CLIJSONDecodeError("{invalid json}", e)
assert error.line == "{invalid json}"
assert error.original_error == e
assert "Failed to decode JSON" in str(error)

191
tests/test_integration.py Normal file
View file

@ -0,0 +1,191 @@
"""Integration tests for Claude SDK.
These tests verify end-to-end functionality with mocked CLI responses.
"""
from unittest.mock import AsyncMock, patch
import anyio
import pytest
from claude_code_sdk import (
AssistantMessage,
ClaudeCodeOptions,
CLINotFoundError,
ResultMessage,
query,
)
from claude_code_sdk.types import ToolUseBlock
class TestIntegration:
"""End-to-end integration tests."""
def test_simple_query_response(self):
"""Test a simple query with text response."""
async def _test():
with patch("claude_code_sdk._internal.client.SubprocessCLITransport") as mock_transport_class:
mock_transport = AsyncMock()
mock_transport_class.return_value = mock_transport
# Mock the message stream
async def mock_receive():
yield {
"type": "assistant",
"message": {
"role": "assistant",
"content": [{"type": "text", "text": "2 + 2 equals 4"}],
},
}
yield {
"type": "result",
"subtype": "success",
"cost_usd": 0.001,
"duration_ms": 1000,
"duration_api_ms": 800,
"is_error": False,
"num_turns": 1,
"session_id": "test-session",
"total_cost": 0.001,
}
mock_transport.receive_messages = mock_receive
mock_transport.connect = AsyncMock()
mock_transport.disconnect = AsyncMock()
# Run query
messages = []
async for msg in query(prompt="What is 2 + 2?"):
messages.append(msg)
# Verify results
assert len(messages) == 2
# Check assistant message
assert isinstance(messages[0], AssistantMessage)
assert len(messages[0].content) == 1
assert messages[0].content[0].text == "2 + 2 equals 4"
# Check result message
assert isinstance(messages[1], ResultMessage)
assert messages[1].cost_usd == 0.001
assert messages[1].session_id == "test-session"
anyio.run(_test)
def test_query_with_tool_use(self):
"""Test query that uses tools."""
async def _test():
with patch("claude_code_sdk._internal.client.SubprocessCLITransport") as mock_transport_class:
mock_transport = AsyncMock()
mock_transport_class.return_value = mock_transport
# Mock the message stream with tool use
async def mock_receive():
yield {
"type": "assistant",
"message": {
"role": "assistant",
"content": [
{
"type": "text",
"text": "Let me read that file for you.",
},
{
"type": "tool_use",
"id": "tool-123",
"name": "Read",
"input": {"file_path": "/test.txt"},
},
],
},
}
yield {
"type": "result",
"subtype": "success",
"cost_usd": 0.002,
"duration_ms": 1500,
"duration_api_ms": 1200,
"is_error": False,
"num_turns": 1,
"session_id": "test-session-2",
"total_cost": 0.002,
}
mock_transport.receive_messages = mock_receive
mock_transport.connect = AsyncMock()
mock_transport.disconnect = AsyncMock()
# Run query with tools enabled
messages = []
async for msg in query(
prompt="Read /test.txt",
options=ClaudeCodeOptions(allowed_tools=["Read"]),
):
messages.append(msg)
# Verify results
assert len(messages) == 2
# Check assistant message with tool use
assert isinstance(messages[0], AssistantMessage)
assert len(messages[0].content) == 2
assert messages[0].content[0].text == "Let me read that file for you."
assert isinstance(messages[0].content[1], ToolUseBlock)
assert messages[0].content[1].name == "Read"
assert messages[0].content[1].input["file_path"] == "/test.txt"
anyio.run(_test)
def test_cli_not_found(self):
"""Test handling when CLI is not found."""
async def _test():
with patch("shutil.which", return_value=None), patch(
"pathlib.Path.exists", return_value=False
), pytest.raises(CLINotFoundError) as exc_info:
async for _ in query(prompt="test"):
pass
assert "Claude Code requires Node.js" in str(exc_info.value)
anyio.run(_test)
def test_continuation_option(self):
"""Test query with continue_conversation option."""
async def _test():
with patch("claude_code_sdk._internal.client.SubprocessCLITransport") as mock_transport_class:
mock_transport = AsyncMock()
mock_transport_class.return_value = mock_transport
# Mock the message stream
async def mock_receive():
yield {
"type": "assistant",
"message": {
"role": "assistant",
"content": [
{
"type": "text",
"text": "Continuing from previous conversation",
}
],
},
}
mock_transport.receive_messages = mock_receive
mock_transport.connect = AsyncMock()
mock_transport.disconnect = AsyncMock()
# Run query with continuation
messages = []
async for msg in query(
prompt="Continue", options=ClaudeCodeOptions(continue_conversation=True)
):
messages.append(msg)
# Verify transport was created with continuation option
mock_transport_class.assert_called_once()
call_kwargs = mock_transport_class.call_args.kwargs
assert call_kwargs['options'].continue_conversation is True
anyio.run(_test)

129
tests/test_transport.py Normal file
View file

@ -0,0 +1,129 @@
"""Tests for Claude SDK transport layer."""
from unittest.mock import AsyncMock, MagicMock, patch
import anyio
import pytest
from claude_code_sdk._internal.transport.subprocess_cli import SubprocessCLITransport
from claude_code_sdk.types import ClaudeCodeOptions
class TestSubprocessCLITransport:
"""Test subprocess transport implementation."""
def test_find_cli_not_found(self):
"""Test CLI not found error."""
from claude_code_sdk._errors import CLINotFoundError
with patch("shutil.which", return_value=None), patch(
"pathlib.Path.exists", return_value=False
), pytest.raises(CLINotFoundError) as exc_info:
SubprocessCLITransport(
prompt="test", options=ClaudeCodeOptions()
)
assert "Claude Code requires Node.js" in str(exc_info.value)
def test_build_command_basic(self):
"""Test building basic CLI command."""
transport = SubprocessCLITransport(
prompt="Hello", options=ClaudeCodeOptions(), cli_path="/usr/bin/claude"
)
cmd = transport._build_command()
assert cmd[0] == "/usr/bin/claude"
assert "--output-format" in cmd
assert "stream-json" in cmd
assert "--print" in cmd
assert "Hello" in cmd
def test_cli_path_accepts_pathlib_path(self):
"""Test that cli_path accepts pathlib.Path objects."""
from pathlib import Path
transport = SubprocessCLITransport(
prompt="Hello", options=ClaudeCodeOptions(), cli_path=Path("/usr/bin/claude")
)
assert transport._cli_path == "/usr/bin/claude"
def test_build_command_with_options(self):
"""Test building CLI command with options."""
transport = SubprocessCLITransport(
prompt="test",
options=ClaudeCodeOptions(
system_prompt="Be helpful",
allowed_tools=["Read", "Write"],
disallowed_tools=["Bash"],
model="claude-3-5-sonnet",
permission_mode="acceptEdits",
max_turns=5,
),
cli_path="/usr/bin/claude",
)
cmd = transport._build_command()
assert "--system-prompt" in cmd
assert "Be helpful" in cmd
assert "--allowedTools" in cmd
assert "Read,Write" in cmd
assert "--disallowedTools" in cmd
assert "Bash" in cmd
assert "--model" in cmd
assert "claude-3-5-sonnet" in cmd
assert "--permission-mode" in cmd
assert "acceptEdits" in cmd
assert "--max-turns" in cmd
assert "5" in cmd
def test_session_continuation(self):
"""Test session continuation options."""
transport = SubprocessCLITransport(
prompt="Continue from before",
options=ClaudeCodeOptions(continue_conversation=True, resume="session-123"),
cli_path="/usr/bin/claude",
)
cmd = transport._build_command()
assert "--continue" in cmd
assert "--resume" in cmd
assert "session-123" in cmd
def test_connect_disconnect(self):
"""Test connect and disconnect lifecycle."""
async def _test():
with patch("anyio.open_process") as mock_exec:
mock_process = MagicMock()
mock_process.returncode = None
mock_process.terminate = MagicMock()
mock_process.wait = AsyncMock()
mock_process.stdout = MagicMock()
mock_process.stderr = MagicMock()
mock_exec.return_value = mock_process
transport = SubprocessCLITransport(
prompt="test", options=ClaudeCodeOptions(), cli_path="/usr/bin/claude"
)
await transport.connect()
assert transport._process is not None
assert transport.is_connected()
await transport.disconnect()
mock_process.terminate.assert_called_once()
anyio.run(_test)
def test_receive_messages(self):
"""Test parsing messages from CLI output."""
# This test is simplified to just test the parsing logic
# The full async stream handling is tested in integration tests
transport = SubprocessCLITransport(
prompt="test", options=ClaudeCodeOptions(), cli_path="/usr/bin/claude"
)
# The actual message parsing is done by the client, not the transport
# So we just verify the transport can be created and basic structure is correct
assert transport._prompt == "test"
assert transport._cli_path == "/usr/bin/claude"

117
tests/test_types.py Normal file
View file

@ -0,0 +1,117 @@
"""Tests for Claude SDK type definitions."""
from claude_code_sdk import (
AssistantMessage,
ClaudeCodeOptions,
ResultMessage,
)
from claude_code_sdk.types import TextBlock, ToolResultBlock, ToolUseBlock, UserMessage
class TestMessageTypes:
"""Test message type creation and validation."""
def test_user_message_creation(self):
"""Test creating a UserMessage."""
msg = UserMessage(content="Hello, Claude!")
assert msg.content == "Hello, Claude!"
def test_assistant_message_with_text(self):
"""Test creating an AssistantMessage with text content."""
text_block = TextBlock(text="Hello, human!")
msg = AssistantMessage(content=[text_block])
assert len(msg.content) == 1
assert msg.content[0].text == "Hello, human!"
def test_tool_use_block(self):
"""Test creating a ToolUseBlock."""
block = ToolUseBlock(
id="tool-123",
name="Read",
input={"file_path": "/test.txt"}
)
assert block.id == "tool-123"
assert block.name == "Read"
assert block.input["file_path"] == "/test.txt"
def test_tool_result_block(self):
"""Test creating a ToolResultBlock."""
block = ToolResultBlock(
tool_use_id="tool-123",
content="File contents here",
is_error=False
)
assert block.tool_use_id == "tool-123"
assert block.content == "File contents here"
assert block.is_error is False
def test_result_message(self):
"""Test creating a ResultMessage."""
msg = ResultMessage(
subtype="success",
cost_usd=0.01,
duration_ms=1500,
duration_api_ms=1200,
is_error=False,
num_turns=1,
session_id="session-123",
total_cost_usd=0.01
)
assert msg.subtype == "success"
assert msg.cost_usd == 0.01
assert msg.session_id == "session-123"
class TestOptions:
"""Test Options configuration."""
def test_default_options(self):
"""Test Options with default values."""
options = ClaudeCodeOptions()
assert options.allowed_tools == []
assert options.max_thinking_tokens == 8000
assert options.system_prompt is None
assert options.permission_mode is None
assert options.continue_conversation is False
assert options.disallowed_tools == []
def test_claude_code_options_with_tools(self):
"""Test Options with built-in tools."""
options = ClaudeCodeOptions(
allowed_tools=["Read", "Write", "Edit"],
disallowed_tools=["Bash"]
)
assert options.allowed_tools == ["Read", "Write", "Edit"]
assert options.disallowed_tools == ["Bash"]
def test_claude_code_options_with_permission_mode(self):
"""Test Options with permission mode."""
options = ClaudeCodeOptions(permission_mode="bypassPermissions")
assert options.permission_mode == "bypassPermissions"
def test_claude_code_options_with_system_prompt(self):
"""Test Options with system prompt."""
options = ClaudeCodeOptions(
system_prompt="You are a helpful assistant.",
append_system_prompt="Be concise."
)
assert options.system_prompt == "You are a helpful assistant."
assert options.append_system_prompt == "Be concise."
def test_claude_code_options_with_session_continuation(self):
"""Test Options with session continuation."""
options = ClaudeCodeOptions(
continue_conversation=True,
resume="session-123"
)
assert options.continue_conversation is True
assert options.resume == "session-123"
def test_claude_code_options_with_model_specification(self):
"""Test Options with model specification."""
options = ClaudeCodeOptions(
model="claude-3-5-sonnet-20241022",
permission_prompt_tool_name="CustomTool"
)
assert options.model == "claude-3-5-sonnet-20241022"
assert options.permission_prompt_tool_name == "CustomTool"