tessl/pypi-claude-code-sdk

Python SDK for Claude Code providing simple query functions and advanced bidirectional interactive conversations with custom tool support

Overview

Eval results

Files

Simple Queries

Name: tessl/pypi-claude-code-sdk
Author: tessl

The query() function provides unidirectional query interface for one-shot interactions with Claude Code. This is ideal for simple, stateless queries where you don't need bidirectional communication or conversation management.

Capabilities

Main Query Function

Sends a prompt to Claude Code and returns an async iterator of response messages. Supports both string prompts for simple queries and async iterables for streaming mode.

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: ClaudeCodeOptions | None = None,
    transport: Transport | None = None,
) -> AsyncIterator[Message]:
    """
    Query Claude Code for one-shot or unidirectional streaming interactions.

    Args:
        prompt: The prompt to send to Claude. Can be a string for single-shot queries
                or an AsyncIterable[dict] for streaming mode with continuous interaction.
        options: Optional configuration (defaults to ClaudeCodeOptions() if None).
        transport: Optional transport implementation. If provided, this will be used
                  instead of the default transport selection based on options.

    Yields:
        Messages from the conversation

    Returns:
        AsyncIterator[Message]: Stream of messages from Claude
    """

Usage Examples

Simple String Query

import anyio
from claude_code_sdk import query

async def main():
    async for message in query(prompt="What is the capital of France?"):
        print(message)

anyio.run(main)

Query with Options

from claude_code_sdk import query, ClaudeCodeOptions, AssistantMessage, TextBlock

async def main():
    options = ClaudeCodeOptions(
        system_prompt="You are an expert Python developer",
        max_turns=1,
        cwd="/home/user/project"
    )

    async for message in query(
        prompt="Create a Python web server",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)

Tool Usage Query

from claude_code_sdk import query, ClaudeCodeOptions

async def main():
    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
        print(message)

anyio.run(main)

Streaming Mode Query

from claude_code_sdk import query

async def main():
    async def prompts():
        yield {"type": "user", "message": {"role": "user", "content": "Hello"}}
        yield {"type": "user", "message": {"role": "user", "content": "How are you?"}}

    # All prompts are sent, then all responses received
    async for message in query(prompt=prompts()):
        print(message)

anyio.run(main)

Custom Transport Query

from claude_code_sdk import query, Transport

class MyCustomTransport(Transport):
    # Implement custom transport logic
    async def connect(self) -> None:
        pass

    async def write(self, data: str) -> None:
        pass

    def read_messages(self):
        # Custom implementation
        yield {}

    # ... implement other required methods

async def main():
    transport = MyCustomTransport()
    async for message in query(
        prompt="Hello",
        transport=transport
    ):
        print(message)