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)

from claude_code_sdk import query_with_oauth

async for message in query_with_oauth(prompt="Hello Claude!"):
    print(message)

from claude_code_sdk import query

async for message in query(prompt="Hello Claude!"):
    print(message)

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)

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

# Method 1: Using session IDs
session_id = None
async for message in query(prompt="Remember my name is Alice"):
    if isinstance(message, ResultMessage):
        session_id = message.session_id

# Continue the conversation
async for message in query(
    prompt="What's my name?",
    options=ClaudeCodeOptions(resume=session_id)
):
    print(message)

# Method 2: Auto-continue last conversation
async for message in query(
    prompt="What did we just discuss?",
    options=ClaudeCodeOptions(continue_conversation=True)
):
    print(message)

from claude_code_sdk.conversation_manager import ConversationManager
from claude_code_sdk.conversation_templates import TemplateManager
from claude_code_sdk.conversation_chains import create_debugging_chain

# Managed conversations with persistence
manager = ConversationManager()
session_id, messages = await manager.create_conversation(
    initial_prompt="Let's build a web app",
    tags=["project", "web"]
)

# Use pre-built templates
template_mgr = TemplateManager()
template = template_mgr.get_template("code_review")

# Automated workflows with chains
debug_chain = create_debugging_chain()
result = await debug_chain.execute(
    context_overrides={"issue_description": "Memory leak"}
)

options = ClaudeCodeOptions(
    # Add additional directories for tool access
    add_dirs=["/path/to/libs", "/path/to/data"],
    
    # Skip all permission prompts (use with caution!)
    dangerously_skip_permissions=True,
    
    # Enable debug logging in CLI
    debug=True,
    
    # Control output verbosity (default: True for SDK)
    verbose=False,
    
    # Specify output format (default: "stream-json")
    output_format="json",  # "text", "json", or "stream-json"
    
    # Specify input format
    input_format="text",  # "text" or "stream-json"
)

from pathlib import Path

options = ClaudeCodeOptions(
    cwd="/path/to/project"  # or Path("/path/to/project")
)

result = await update()
print(result)

# List MCP servers
servers = await mcp(["list"])

# Add MCP server
result = await mcp(["add", "my-server"])

# Remove MCP server
result = await mcp(["remove", "my-server"])

# Get configuration value
model = await config(["get", "model"])

# Set configuration value
result = await config(["set", "model", "claude-opus-4"])

# List all config
all_config = await config(["list"])

health = await doctor()
print(health)

ver = await version()
print(f"Claude Code version: {ver}")

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}")

try:
    async for message in query(prompt="test"):
        pass
except ProcessError as e:
    print(f"Exit code: {e.exit_code}")
    print(f"Error details: {e.stderr}")

import asyncio

async def long_operation():
    async with asyncio.timeout(300):  # 5 minutes
        async for message in query(prompt="Complex task..."):
            # Process messages
            pass

# Use a different port
server.run(host="0.0.0.0", port=8080)

# Check if port is in use
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex(('localhost', 8000))
if result == 0:
    print("Port 8000 is already in use")

import logging

# Enable debug logging
logging.basicConfig(level=logging.DEBUG)

# The SDK will now log:
# - CLI discovery process
# - Connection attempts
# - Message parsing
# - Error details

import claude_code_sdk
print(f"SDK version: {claude_code_sdk.__version__}")

async for message in query(prompt="Debug test"):
    print(f"Message type: {type(message).__name__}")
    print(f"Message content: {message}")
    
    if hasattr(message, 'content'):
        for block in message.content:
            print(f"  Block type: {type(block).__name__}")

# Start with the simplest possible query
async for message in query(prompt="Hello"):
    print(message)

# If that works, gradually add complexity
options = ClaudeCodeOptions(allowed_tools=["Read"])
async for message in query(prompt="Read README.md", options=options):
    print(message)

# Good: Automatic cleanup
async def process_files():
    async for message in query(prompt="Process files"):
        # SDK handles cleanup automatically
        pass

# For long-running applications
import asyncio

async def main():
    try:
        async for message in query(prompt="Long task"):
            process_message(message)
    except KeyboardInterrupt:
        # Graceful shutdown
        print("Shutting down...")

from claude_code_sdk import (
    query, 
    CLINotFoundError,
    ProcessError,
    CLIConnectionError
)

async def safe_query(prompt: str, max_retries: int = 3):
    """Query with automatic retry and error handling."""
    for attempt in range(max_retries):
        try:
            async for message in query(prompt=prompt):
                yield message
            break
        except CLINotFoundError:
            # Can't recover from this
            raise
        except (CLIConnectionError, ProcessError) as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)
                continue
            raise

# Good: Only request needed tools
options = ClaudeCodeOptions(
    allowed_tools=["Read", "Write"],  # Specific tools
    disallowed_tools=["Bash"],        # Explicitly disallow
)

# Better: Use appropriate permission mode
options = ClaudeCodeOptions(
    allowed_tools=["Read", "Write", "Edit"],
    permission_mode="acceptEdits",  # Auto-accept safe operations
)

# Avoid: Too permissive
# options = ClaudeCodeOptions(permission_mode="bypassPermissions")

from claude_code_sdk import AssistantMessage, TextBlock, ToolUseBlock

async def process_claude_response(prompt: str):
    """Process different message types appropriately."""
    tool_uses = []
    text_responses = []
    
    async for message in query(prompt=prompt):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    text_responses.append(block.text)
                elif isinstance(block, ToolUseBlock):
                    tool_uses.append({
                        "tool": block.name,
                        "input": block.input
                    })
    
    return {
        "text": "\n".join(text_responses),
        "tools_used": tool_uses
    }

# Continue previous session
options = ClaudeCodeOptions(continue_conversation=True)

# Or save and resume specific sessions
session_id = None

async def chat(user_input: str):
    global session_id
    
    options = ClaudeCodeOptions(
        resume=session_id if session_id else None,
        max_turns=10  # Prevent runaway conversations
    )
    
    async for message in query(prompt=user_input, options=options):
        if isinstance(message, ResultMessage):
            session_id = message.session_id
        yield message

from pathlib import Path

# Good: Explicit, absolute path
options = ClaudeCodeOptions(
    cwd=Path("/home/user/projects/myapp").absolute()
)

# Good: Relative to script location
script_dir = Path(__file__).parent
options = ClaudeCodeOptions(
    cwd=script_dir / "workspace"
)

# Avoid: Implicit current directory
# options = ClaudeCodeOptions()  # Uses wherever script is run from

# Batch operations when possible
prompt = """
Please perform these tasks:
1. Read all Python files in the src/ directory
2. Analyze the code structure
3. Generate a summary report
"""

# Instead of multiple queries
# DON'T: Multiple round trips
# for file in files:
#     async for msg in query(f"Read {file}"):
#         ...

# DO: Single comprehensive request
async for message in query(prompt=prompt, options=options):
    # Process all results at once
    pass

import logging
from claude_code_sdk import ResultMessage

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

logger = logging.getLogger(__name__)

async def monitored_query(prompt: str):
    """Query with monitoring."""
    start_time = time.time()
    
    try:
        async for message in query(prompt=prompt):
            if isinstance(message, ResultMessage):
                logger.info(
                    f"Query completed - "
                    f"Cost: ${message.cost_usd:.4f}, "
                    f"Duration: {message.duration_ms}ms, "
                    f"Session: {message.session_id}"
                )
            yield message
    except Exception as e:
        logger.error(f"Query failed: {e}", exc_info=True)
        raise
    finally:
        duration = time.time() - start_time
        logger.info(f"Total duration: {duration:.2f}s")

import os

# Good: Use environment variables
options = ClaudeCodeOptions(
    system_prompt="Use the API key from environment"
)

# Set via environment
os.environ["ANTHROPIC_API_KEY"] = "your-key"

# Avoid: Hardcoding secrets
# DON'T: options = ClaudeCodeOptions(
#     system_prompt="Use API key: sk-ant-..."
# )

# Good: Sanitize file paths
safe_path = Path(user_input).resolve()
if not safe_path.is_relative_to(allowed_directory):
    raise ValueError("Access denied")

# Make your code testable
async def analyze_code(
    file_path: str,
    query_func=query  # Injectable for testing
):
    """Analyze code with injectable query function."""
    prompt = f"Analyze the code in {file_path}"
    
    async for message in query_func(prompt=prompt):
        # Process message
        pass

# In tests
async def mock_query(prompt, options=None):
    """Mock query for testing."""
    yield AssistantMessage(content=[
        TextBlock(text="Mock analysis complete")
    ])

# Test
async def test_analyze():
    await analyze_code("test.py", query_func=mock_query)

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def timeout_query(prompt: str, timeout_seconds: int = 300):
    """Query with timeout and cleanup."""
    task = None
    try:
        async with asyncio.timeout(timeout_seconds):
            task = asyncio.create_task(collect_messages(prompt))
            yield task
    except asyncio.TimeoutError:
        logger.warning(f"Query timed out after {timeout_seconds}s")
        raise
    finally:
        if task and not task.done():
            task.cancel()

async def collect_messages(prompt: str):
    """Collect all messages from a query."""
    messages = []
    async for message in query(prompt=prompt):
        messages.append(message)
    return messages

from claude_code_sdk.websocket_server import EnhancedClaudeWebSocketServer

server = EnhancedClaudeWebSocketServer()
server.run(host="0.0.0.0", port=8000)

from claude_code_sdk import query, ClaudeCodeOptions
from claude_code_agent_system import BaseAgent

class CustomAgent(BaseAgent):
    async def process_with_claude(self, prompt: str):
        options = ClaudeCodeOptions(allowed_tools=["Read", "Write"])
        async for message in query(prompt=prompt, options=options):
            # Process responses
            pass

from claude_code_sdk.notebook_utils import display_claude_response, render_markdown

# Beautiful markdown rendering
async for message in query(prompt="Explain Python decorators"):
    display_claude_response(message)

# Or render markdown directly
render_markdown("# Hello\nThis is **bold** text with `code`")