Simple Query Interface

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

    This function is ideal for simple, stateless queries where you don't need
    bidirectional communication or conversation management. For interactive,
    stateful conversations, use ClaudeSDKClient instead.

    Key differences from ClaudeSDKClient:
    - Unidirectional: Send all messages upfront, receive all responses
    - Stateless: Each query is independent, no conversation state
    - Simple: Fire-and-forget style, no connection management
    - No interrupts: Cannot interrupt or send follow-up messages

    When to use query():
    - Simple one-off questions
    - Batch processing of independent prompts
    - Code generation or analysis tasks
    - Automated scripts and CI/CD pipelines
    - When you know all inputs upfront

    When to use ClaudeSDKClient:
    - Interactive conversations with follow-ups
    - Chat applications or REPL-like interfaces
    - When you need to send messages based on responses
    - When you need interrupt capabilities
    - Long-running sessions with state

    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.
                In streaming mode, each dict should have the structure:
                {
                    "type": "user",
                    "message": {"role": "user", "content": "..."},
                    "parent_tool_use_id": None,
                    "session_id": "..."
                }
        options: Optional configuration (defaults to ClaudeAgentOptions() 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.
        transport: Optional transport implementation. If provided, this will be used
                  instead of the default transport selection based on options.

    Yields:
        Message objects from the conversation (UserMessage, AssistantMessage,
        SystemMessage, ResultMessage, or StreamEvent)
    """

import anyio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="What is 2 + 2?"):
        print(message)

anyio.run(main)

from claude_agent_sdk import query, AssistantMessage, TextBlock, ResultMessage

async def main():
    async for message in query(prompt="What is the capital of France?"):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(f"Claude: {block.text}")
        elif isinstance(message, ResultMessage):
            print(f"Cost: ${message.total_cost_usd:.4f}")
            print(f"Duration: {message.duration_ms}ms")

anyio.run(main)

from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful Python expert",
        max_turns=1,
        cwd="/home/user/project"
    )

    async for message in query(
        prompt="Explain what asyncio is",
        options=options
    ):
        print(message)

anyio.run(main)

from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode='acceptEdits'  # auto-accept file edits
    )

    async for message in query(
        prompt="Create a hello.py file that prints 'Hello, World!'",
        options=options
    ):
        # Process tool use and results
        pass

anyio.run(main)

from pathlib import Path
from claude_agent_sdk import query, ClaudeAgentOptions

async def main():
    options = ClaudeAgentOptions(
        cwd=Path("/path/to/project")  # or "/path/to/project" as string
    )

    async for message in query(
        prompt="List the files in the current directory",
        options=options
    ):
        print(message)

anyio.run(main)

from claude_agent_sdk import query

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

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

anyio.run(main)

from claude_agent_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):
        pass

    async def close(self) -> None:
        pass

    def is_ready(self) -> bool:
        return True

    async def end_input(self) -> None:
        pass

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

anyio.run(main)

from claude_agent_sdk import (
    query,
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError
)

async def main():
    try:
        async for message in query(prompt="Hello Claude"):
            print(message)
    except CLINotFoundError:
        print("Please install Claude Code: npm install -g @anthropic-ai/claude-code")
    except CLIConnectionError as e:
        print(f"Connection error: {e}")
    except ProcessError as e:
        print(f"Process failed with exit code {e.exit_code}: {e.stderr}")
    except CLIJSONDecodeError as e:
        print(f"JSON decode error: {e.line}")
    except ClaudeSDKError as e:
        print(f"SDK error: {e}")

anyio.run(main)