Claude Agent SDK for Python (anthropics/claude-agent-sdk-python)

Claude Agent SDK for Python

https://github.com/anthropics/claude-agent-sdk-python
Admin
A Python SDK that provides functionalities for interacting with Claude Agent, enabling querying,...

Tokens:15,505
Snippets:46
Trust Score:8.8
Update:1 week ago
Show doc for...
Context Summary (auto-generated)
Raw
# Claude Agent SDK for Python

The Claude Agent SDK for Python is a comprehensive library for building AI agents powered by Claude. It provides both simple one-shot query capabilities and full bidirectional streaming conversations with Claude Code, enabling developers to create intelligent automation tools, coding assistants, and interactive AI applications. The SDK bundles the Claude Code CLI automatically, requiring no separate installation.

The SDK offers two primary interfaces: `query()` for simple, stateless interactions where all inputs are known upfront, and `ClaudeSDKClient` for interactive, multi-turn conversations with support for custom tools (via in-process MCP servers), hooks for intercepting and controlling agent behavior, real-time streaming, and session management. It supports Python 3.10+ and works with any async runtime (asyncio, trio, anyio).

## Installation

```bash
pip install claude-agent-sdk
```

## query() - Simple One-Shot Queries

The `query()` function provides a simple async iterator interface for one-shot interactions with Claude. It's ideal for stateless queries where you don't need bidirectional communication.

```python
import anyio
from claude_agent_sdk import (
    query,
    ClaudeAgentOptions,
    AssistantMessage,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
)

async def main():
    # Simple query - basic question
    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}")
        elif isinstance(message, ResultMessage):
            print(f"Total cost: ${message.total_cost_usd:.4f}")

    # Query with custom options
    options = ClaudeAgentOptions(
        system_prompt="You are a helpful coding assistant.",
        max_turns=3,
        allowed_tools=["Read", "Write", "Bash"],
        permission_mode="acceptEdits",  # Auto-accept file edits
        cwd="/path/to/project",
    )

    async for message in query(
        prompt="Create a hello.py file that prints 'Hello World'",
        options=options
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)
                elif isinstance(block, ToolUseBlock):
                    print(f"Using tool: {block.name} with input: {block.input}")

anyio.run(main)
```

## ClaudeSDKClient - Interactive Conversations

`ClaudeSDKClient` provides a full bidirectional streaming interface for interactive conversations. It supports multi-turn conversations, custom tools, hooks, interrupts, and session management.

```python
import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    UserMessage,
)

async def main():
    # Basic streaming conversation
    async with ClaudeSDKClient() as client:
        # First turn
        await client.query("What's the capital of France?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # Second turn - follow-up question (maintains context)
        await client.query("What's the population of that city?")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")
            elif isinstance(msg, ResultMessage):
                print(f"Session ID: {msg.session_id}")
                print(f"Cost: ${msg.total_cost_usd:.4f}")

    # With custom options
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        system_prompt="You are a senior software engineer.",
        max_budget_usd=1.0,  # Maximum spending limit
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files in the current directory")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)
                    elif isinstance(block, ToolUseBlock):
                        print(f"Tool: {block.name}, Input: {block.input}")

asyncio.run(main)
```

## Custom Tools with SDK MCP Servers

Create in-process MCP servers that run directly within your Python application. This provides better performance than external MCP servers by eliminating IPC overhead.

```python
import asyncio
from typing import Any
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    create_sdk_mcp_server,
    tool,
    AssistantMessage,
    TextBlock,
    ToolUseBlock,
)

# Define tools using the @tool decorator
@tool("add", "Add two numbers together", {"a": float, "b": float})
async def add_numbers(args: dict[str, Any]) -> dict[str, Any]:
    result = args["a"] + args["b"]
    return {"content": [{"type": "text", "text": f"{args['a']} + {args['b']} = {result}"}]}

@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply_numbers(args: dict[str, Any]) -> dict[str, Any]:
    result = args["a"] * args["b"]
    return {"content": [{"type": "text", "text": f"{args['a']} x {args['b']} = {result}"}]}

@tool("divide", "Divide two numbers", {"a": float, "b": float})
async def divide_numbers(args: dict[str, Any]) -> dict[str, Any]:
    if args["b"] == 0:
        return {
            "content": [{"type": "text", "text": "Error: Division by zero"}],
            "is_error": True,
        }
    result = args["a"] / args["b"]
    return {"content": [{"type": "text", "text": f"{args['a']} / {args['b']} = {result}"}]}

async def main():
    # Create SDK MCP server with tools
    calculator = create_sdk_mcp_server(
        name="calculator",
        version="1.0.0",
        tools=[add_numbers, multiply_numbers, divide_numbers],
    )

    # Configure Claude to use the calculator server
    options = ClaudeAgentOptions(
        mcp_servers={"calc": calculator},
        # Pre-approve tools so they run without permission prompts
        allowed_tools=[
            "mcp__calc__add",
            "mcp__calc__multiply",
            "mcp__calc__divide",
        ],
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Calculate 15 + 27, then multiply the result by 3")

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")
                    elif isinstance(block, ToolUseBlock):
                        print(f"Using: {block.name}({block.input})")

asyncio.run(main)
```

## Hooks - Intercept and Control Agent Behavior

Hooks allow you to intercept tool calls, modify inputs/outputs, add context, and control execution flow. They're Python functions invoked at specific points in the Claude agent loop.

```python
import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    ResultMessage,
    TextBlock,
)
from claude_agent_sdk.types import (
    HookContext,
    HookInput,
    HookJSONOutput,
    HookMatcher,
)

# PreToolUse hook - intercept before tool execution
async def security_check_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]

    # Block dangerous bash commands
    if tool_name == "Bash":
        command = tool_input.get("command", "")
        dangerous_patterns = ["rm -rf", "sudo", "chmod 777", "> /dev/"]

        for pattern in dangerous_patterns:
            if pattern in command:
                return {
                    "reason": f"Security policy blocks commands containing: {pattern}",
                    "hookSpecificOutput": {
                        "hookEventName": "PreToolUse",
                        "permissionDecision": "deny",
                        "permissionDecisionReason": f"Blocked dangerous pattern: {pattern}",
                    }
                }

    # Block writes to sensitive files
    if tool_name == "Write":
        file_path = tool_input.get("file_path", "")
        if any(s in file_path for s in [".env", "secrets", "credentials", "password"]):
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": "Cannot write to sensitive files",
                }
            }

    # Allow everything else
    return {
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "allow",
        }
    }

# PostToolUse hook - process after tool execution
async def audit_tool_output(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    tool_response = input_data.get("tool_response", "")

    # Add warning if tool produced an error
    if "error" in str(tool_response).lower():
        return {
            "systemMessage": "Tool execution produced an error",
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "additionalContext": "Consider trying an alternative approach.",
            }
        }
    return {}

# UserPromptSubmit hook - add context to every prompt
async def add_context_hook(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": "Remember: Always write clean, well-documented code.",
        }
    }

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Bash", "Write", "Read"],
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="Bash|Write", hooks=[security_check_hook]),
            ],
            "PostToolUse": [
                HookMatcher(matcher=None, hooks=[audit_tool_output]),
            ],
            "UserPromptSubmit": [
                HookMatcher(matcher=None, hooks=[add_context_hook]),
            ],
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        # This will be blocked by the security hook
        await client.query("Run the command: rm -rf /tmp/test")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # This will be allowed
        await client.query("Run the command: echo 'Hello World'")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

asyncio.run(main)
```

## Permission Callbacks - Dynamic Tool Approval

Use the `can_use_tool` callback for programmatic control over tool permissions at runtime.

```python
import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)
from claude_agent_sdk.types import (
    CanUseTool,
    ToolPermissionContext,
    PermissionResult,
    PermissionResultAllow,
    PermissionResultDeny,
)

# Approved paths for file operations
ALLOWED_PATHS = ["/tmp/", "/home/user/projects/"]

async def permission_callback(
    tool_name: str,
    tool_input: dict,
    context: ToolPermissionContext
) -> PermissionResult:
    """Dynamically approve or deny tool usage."""

    if tool_name == "Write":
        file_path = tool_input.get("file_path", "")

        # Check if path is in allowed directories
        if any(file_path.startswith(p) for p in ALLOWED_PATHS):
            return PermissionResultAllow(
                behavior="allow",
                updated_input=None,  # Can modify input if needed
            )
        else:
            return PermissionResultDeny(
                behavior="deny",
                message=f"Writing to {file_path} is not allowed. Only /tmp/ and /home/user/projects/ are permitted.",
                interrupt=False,  # Set True to stop the conversation
            )

    if tool_name == "Bash":
        command = tool_input.get("command", "")

        # Allow read-only commands
        read_only = ["ls", "cat", "head", "tail", "grep", "find", "echo"]
        if any(command.strip().startswith(cmd) for cmd in read_only):
            return PermissionResultAllow(behavior="allow")

        return PermissionResultDeny(
            behavior="deny",
            message="Only read-only bash commands are allowed.",
        )

    # Allow all other tools
    return PermissionResultAllow(behavior="allow")

async def main():
    options = ClaudeAgentOptions(
        can_use_tool=permission_callback,
        allowed_tools=["Read"],  # Read is pre-approved, others go through callback
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("List files in /tmp and write 'test' to /tmp/test.txt")

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

asyncio.run(main)
```

## Session Management

List, retrieve, and manage conversation sessions stored locally.

```python
from claude_agent_sdk import (
    list_sessions,
    get_session_info,
    get_session_messages,
    rename_session,
    tag_session,
    delete_session,
    fork_session,
    list_subagents,
    get_subagent_messages,
)

# List all sessions for a project
sessions = list_sessions(
    directory="/path/to/project",
    limit=20,
    offset=0,
    include_worktrees=True,  # Include git worktree sessions
)

for session in sessions:
    print(f"Session: {session.session_id}")
    print(f"  Summary: {session.summary}")
    print(f"  Title: {session.custom_title}")
    print(f"  First prompt: {session.first_prompt}")
    print(f"  Git branch: {session.git_branch}")
    print(f"  Tag: {session.tag}")
    print(f"  Last modified: {session.last_modified}")
    print(f"  Created at: {session.created_at}")

# Get info for a specific session
info = get_session_info(
    "550e8400-e29b-41d4-a716-446655440000",
    directory="/path/to/project",
)
if info:
    print(f"Session summary: {info.summary}")

# Read conversation messages from a session
messages = get_session_messages(
    session_id="550e8400-e29b-41d4-a716-446655440000",
    directory="/path/to/project",
    limit=50,
    offset=0,
)

for msg in messages:
    print(f"[{msg.type}] {msg.uuid}: {msg.message}")

# Rename a session
rename_session(
    "550e8400-e29b-41d4-a716-446655440000",
    "My refactoring session",
    directory="/path/to/project",
)

# Tag a session for organization
tag_session(
    "550e8400-e29b-41d4-a716-446655440000",
    "experiment",
    directory="/path/to/project",
)

# Clear a tag
tag_session("550e8400-e29b-41d4-a716-446655440000", None)

# Fork a session to create a branch
result = fork_session(
    "550e8400-e29b-41d4-a716-446655440000",
    directory="/path/to/project",
    up_to_message_id="660e8400-e29b-41d4-a716-446655440001",  # Optional: fork from specific point
    title="Fork: trying alternative approach",
)
print(f"Forked to new session: {result.session_id}")

# List subagents spawned during a session
agent_ids = list_subagents(
    "550e8400-e29b-41d4-a716-446655440000",
    directory="/path/to/project",
)
print(f"Subagents: {agent_ids}")

# Read subagent transcript
subagent_messages = get_subagent_messages(
    session_id="550e8400-e29b-41d4-a716-446655440000",
    agent_id="agent-abc123",
    directory="/path/to/project",
)

# Delete a session (permanent!)
delete_session("550e8400-e29b-41d4-a716-446655440000")
```

## Interrupt and Dynamic Control

Interrupt ongoing operations and dynamically change settings during a conversation.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, AssistantMessage, TextBlock

async def main():
    async with ClaudeSDKClient() as client:
        # Start a long-running task
        await client.query("Count from 1 to 100 slowly, pausing between each number")

        # Background task to receive messages
        async def receive():
            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(f"Claude: {block.text[:50]}...")

        receive_task = asyncio.create_task(receive())

        # Wait 2 seconds then interrupt
        await asyncio.sleep(2)
        print("\n[Sending interrupt...]")
        await client.interrupt()

        # Wait for task to complete
        await receive_task

        # Send new instruction after interrupt
        await client.query("Just say 'Hello!'")
        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(f"Claude: {block.text}")

        # Change permission mode dynamically
        await client.set_permission_mode("acceptEdits")

        # Change model during conversation
        await client.set_model("claude-sonnet-4-5")

        # Get server info
        server_info = await client.get_server_info()
        if server_info:
            print(f"Available commands: {len(server_info.get('commands', []))}")

        # Get MCP server status
        mcp_status = await client.get_mcp_status()
        for server in mcp_status.get("mcpServers", []):
            print(f"MCP Server: {server['name']} - {server['status']}")

        # Get context usage
        usage = await client.get_context_usage()
        print(f"Context usage: {usage['percentage']:.1f}%")
        print(f"Total tokens: {usage['totalTokens']} / {usage['maxTokens']}")

asyncio.run(main)
```

## Programmatic Subagents

Define custom subagents inline in code for specialized tasks.

```python
import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)
from claude_agent_sdk.types import AgentDefinition

async def main():
    options = ClaudeAgentOptions(
        agents={
            "code-reviewer": AgentDefinition(
                description="Expert code reviewer for Python projects",
                prompt="You are a senior Python developer. Review code for bugs, performance issues, and style. Be constructive and specific.",
                tools=["Read", "Grep", "Glob"],
                model="claude-sonnet-4-5",
                maxTurns=5,
            ),
            "test-writer": AgentDefinition(
                description="Test case generator",
                prompt="You write comprehensive pytest test cases. Cover edge cases and error conditions.",
                tools=["Read", "Write"],
                disallowedTools=["Bash"],  # Prevent running tests
            ),
        },
        allowed_tools=["Task"],  # Allow spawning subagents
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query(
            "Use the code-reviewer agent to review the files in src/ "
            "and the test-writer agent to write missing tests"
        )

        async for msg in client.receive_response():
            if isinstance(msg, AssistantMessage):
                for block in msg.content:
                    if isinstance(block, TextBlock):
                        print(block.text)

asyncio.run(main)
```

## Structured Output

Get validated JSON responses matching a schema.

```python
import asyncio
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions, ResultMessage

async def main():
    options = ClaudeAgentOptions(
        output_format={
            "type": "json_schema",
            "schema": {
                "type": "object",
                "properties": {
                    "summary": {"type": "string"},
                    "key_points": {
                        "type": "array",
                        "items": {"type": "string"}
                    },
                    "sentiment": {
                        "type": "string",
                        "enum": ["positive", "negative", "neutral"]
                    },
                    "confidence": {"type": "number", "minimum": 0, "maximum": 1}
                },
                "required": ["summary", "key_points", "sentiment", "confidence"]
            }
        }
    )

    async with ClaudeSDKClient(options=options) as client:
        await client.query("Analyze this text: 'The product exceeded my expectations. Great quality and fast shipping!'")

        async for msg in client.receive_response():
            if isinstance(msg, ResultMessage):
                # structured_output contains the validated JSON
                output = msg.structured_output
                print(f"Summary: {output['summary']}")
                print(f"Key points: {output['key_points']}")
                print(f"Sentiment: {output['sentiment']}")
                print(f"Confidence: {output['confidence']}")

asyncio.run(main)
```

## Error Handling

Comprehensive error handling for SDK operations.

```python
import asyncio
from claude_agent_sdk import (
    query,
    ClaudeSDKClient,
    ClaudeAgentOptions,
    ClaudeSDKError,
    CLINotFoundError,
    CLIConnectionError,
    ProcessError,
    CLIJSONDecodeError,
)

async def main():
    try:
        async with ClaudeSDKClient() as client:
            await client.query("Hello")

            try:
                async with asyncio.timeout(5.0):
                    async for msg in client.receive_response():
                        print(msg)
            except asyncio.TimeoutError:
                print("Response timed out after 5 seconds")
                # Gracefully handle timeout

    except CLINotFoundError as e:
        print(f"Claude Code CLI not found: {e}")
        print("Install with: pip install claude-agent-sdk")

    except CLIConnectionError as e:
        print(f"Failed to connect to Claude: {e}")

    except ProcessError as e:
        print(f"CLI process failed: {e}")
        print(f"Exit code: {e.exit_code}")
        print(f"Stderr: {e.stderr}")

    except CLIJSONDecodeError as e:
        print(f"Failed to parse CLI response: {e}")
        print(f"Raw line: {e.line}")

    except ClaudeSDKError as e:
        print(f"SDK error: {e}")

asyncio.run(main)
```

## ClaudeAgentOptions Reference

Complete configuration options for the SDK.

```python
from pathlib import Path
from claude_agent_sdk import ClaudeAgentOptions
from claude_agent_sdk.types import (
    AgentDefinition,
    HookMatcher,
    SandboxSettings,
    ThinkingConfigAdaptive,
)

options = ClaudeAgentOptions(
    # Tool configuration
    tools=["Read", "Write", "Bash"],  # Available tools, or {"type": "preset", "preset": "claude_code"}
    allowed_tools=["Read"],  # Pre-approved tools (no permission prompt)
    disallowed_tools=["Bash"],  # Blocked tools

    # System prompt
    system_prompt="You are a helpful assistant",  # String, or preset dict
    # system_prompt={"type": "preset", "preset": "claude_code", "append": "Extra instructions"}
    # system_prompt={"type": "file", "path": "/path/to/prompt.txt"}

    # Model configuration
    model="claude-sonnet-4-5",
    fallback_model="claude-haiku-3-5",

    # Session management
    session_id="custom-session-id",  # Custom session ID
    resume="existing-session-id",  # Resume existing session
    fork_session=True,  # Fork instead of continue when resuming
    continue_conversation=True,  # Continue previous conversation

    # Limits
    max_turns=10,
    max_budget_usd=5.0,

    # Permission handling
    permission_mode="acceptEdits",  # default, acceptEdits, plan, bypassPermissions, dontAsk, auto
    can_use_tool=my_permission_callback,  # Dynamic permission callback

    # Thinking configuration
    thinking=ThinkingConfigAdaptive(),  # adaptive, enabled (with budget_tokens), or disabled
    effort="high",  # low, medium, high, max

    # MCP servers
    mcp_servers={
        "my-server": my_sdk_server,  # SDK MCP server
        "external": {"type": "stdio", "command": "python", "args": ["-m", "server"]},
    },

    # Hooks
    hooks={
        "PreToolUse": [HookMatcher(matcher="Bash", hooks=[my_hook])],
    },

    # Subagents
    agents={
        "helper": AgentDefinition(description="Helper agent", prompt="Help with tasks"),
    },

    # Working directory and paths
    cwd="/path/to/project",
    add_dirs=["/additional/dir"],
    cli_path="/custom/path/to/claude",

    # Settings
    settings="/path/to/settings.json",
    setting_sources=["user", "project", "local"],

    # Environment
    env={"CUSTOM_VAR": "value"},

    # Streaming options
    include_partial_messages=True,  # Receive partial message updates

    # Sandbox settings
    sandbox=SandboxSettings(
        enabled=True,
        autoAllowBashIfSandboxed=True,
        excludedCommands=["git", "docker"],
    ),

    # Beta features
    betas=["context-1m-2025-08-07"],

    # File checkpointing
    enable_file_checkpointing=True,

    # Output format for structured responses
    output_format={"type": "json_schema", "schema": {...}},

    # Task budget
    task_budget={"total": 50000},  # Token budget

    # Extra CLI arguments
    extra_args={"replay-user-messages": None},
)
```

## Summary

The Claude Agent SDK for Python enables building sophisticated AI agents and automation tools. The primary use cases include: (1) **Simple automation scripts** using `query()` for one-shot tasks like code generation, file processing, and batch operations; (2) **Interactive applications** using `ClaudeSDKClient` for chat interfaces, debugging sessions, and multi-turn conversations; (3) **Tool-augmented agents** with custom MCP servers for domain-specific functionality like calculators, database access, or API integrations; (4) **Controlled execution** with hooks and permission callbacks for enterprise security policies and audit requirements.

Integration patterns include: using the context manager (`async with ClaudeSDKClient()`) for automatic connection management; implementing permission callbacks for dynamic tool approval based on runtime context; creating reusable SDK MCP servers that can be shared across multiple sessions; combining hooks for logging, security, and behavior modification; and leveraging session management for conversation persistence and forking. The SDK's async-first design works seamlessly with asyncio, trio, and anyio, making it suitable for both simple scripts and complex async applications.