Streaming One-Shot Queries

In the previous lesson, you learned the conceptual foundation of the Claude Agent SDK — understanding that Claude Code is the local agent runtime and the SDK is your programmatic interface to control it. Now it's time to write your first actual code that launches an agent and handles its response. You'll learn how to use the query() function to send a prompt, process streaming messages, extract text content, and read valuable metrics about the interaction.

Before diving into agent interactions, you need to understand a key architectural choice in the Claude Agent SDK: it's built on anyio, not the standard asyncio library. This choice affects how you structure your code and run async functions.

If you've worked with Python's built-in asyncio library, you're already familiar with async programming patterns. The Claude Agent SDK uses similar patterns but through anyio, which is a compatibility layer that works with both asyncio and trio. While asyncio is Python's standard async library, anyio provides a cleaner, more consistent API and better handling of subprocesses and streaming—exactly what the SDK needs for managing the Claude Code runtime.

Here's how the two libraries compare in practice:

# Using asyncio (standard Python async)
import asyncio

async def main():
    print("Starting task...")
    await asyncio.sleep(2)
    print("Task complete!")

if __name__ == "__main__":
    asyncio.run(main())

# Using anyio (what the Claude Agent SDK uses)
import anyio

async def main():
    print("Starting task...")
    await anyio.sleep(2)
    print("Task complete!")

if __name__ == "__main__":
    anyio.run(main)

The patterns are nearly identical—you still define async functions, use await for async operations, and run your main function through a runtime. The key differences are:

• anyio.run() instead of asyncio.run() — anyio handles event loop setup automatically
• anyio.sleep() instead of asyncio.sleep() — anyio provides its own async primitives
• Better subprocess handling — anyio excels at managing subprocesses (crucial since the SDK controls Claude Code as a subprocess)

You don't strictly need to use anyio in your own code—you could use pure asyncio if you prefer—but anyio is the path of least resistance since the SDK's async patterns are already designed around it. Throughout this course, all examples will use anyio for consistency, and you'll find it integrates seamlessly with the SDK's streaming APIs.

Understanding this foundation is important because every interaction with the SDK—from simple queries to complex multi-turn sessions—uses async patterns. When you see async for loops iterating over message streams or await statements waiting for agent responses, you're working with anyio's async infrastructure managing the communication between your code and the Claude Code runtime.

Building on your understanding of the Claude Agent SDK's async foundation, the query() function is your simplest entry point for interacting with the Claude Code agent. It allows you to send a single prompt to the agent runtime and stream back everything the agent does in response—all without needing to manage a session or conversation history.

Here's what that looks like in code:

import anyio
from claude_agent_sdk import query

async def main():
    async for message in query(prompt="Hi Claude! Please introduce yourself."):
        # Each message represents one step in the agent's process
        print(message)

if __name__ == "__main__":
    anyio.run(main)

When you use query(), the SDK launches a temporary agent instance, sends your prompt as the only message, and then streams back each step of the agent's reasoning and actions. Here's what happens behind the scenes:

• The SDK starts a short-lived Claude Code agent session using the local runtime.
• Your prompt is delivered as the sole message in this session.
• The agent reasons about the task and plans its approach.
• The agent generates responses based on what it's allowed to do.
  • By default, the agent can see available tools (like Read, Write, or Bash) but cannot execute them without explicit permission configuration. We'll explore tool configuration and permission modes in later lessons.
• Each step is streamed back to you in real time as a sequence of typed messages.
• When the agent finishes, the session closes—no memory or context is kept.

This stateless, one-shot approach makes query() ideal for single, self-contained actions such as "summarize this file," "create a script," or "refactor this function." You simply pass your prompt as a single string, and if you need to provide more context, you add it to that same string. This differs from working directly with the Anthropic API, where you'd build context by passing an array of message objects with roles and content—with query(), you manage all your context within a single string instead. Each call is independent and self-contained, with no conversation state carried between calls. While the function also supports an AsyncIterable for more advanced batching scenarios, we'll focus on simple string prompts for now.

The responses stream back as an async generator that yields message objects as the agent works. Rather than waiting for the entire process to complete, you receive each step as it happens, allowing you to display progress, log actions, or react to specific tool uses in real time. The async for loop waits for each message to arrive, processes it, and then waits for the next one until the agent completes its work. Notice that we're using anyio.run() to execute our async function—the SDK is built on async Python patterns, which allow it to handle streaming efficiently.

When you run the code, you'll see a sequence of messages arrive:

SystemMessage(subtype='init', data={'type': 'system', 'subtype': 'init', 'cwd': '/usercode/FILESYSTEM', 'session_id': 'aab399c5-ff62-4552-a728-8e496fbd67b7', 'tools': ['Task', 'Bash', 'Glob', 'Grep', 'ExitPlanMode', 'Read', 'Edit', 'Write', 'NotebookEdit', 'WebFetch', 'TodoWrite', 'WebSearch', 'BashOutput', 'KillShell', 'Skill', 'SlashCommand'], 'mcp_servers': [], 'model': 'claude-sonnet-4-5-20250929', 'permissionMode': 'default', 'slash_commands': ['compact', 'context', 'cost', 'init', 'pr-comments', 'release-notes', 'todos', 'review', 'security-review'], 'apiKeySource': 'ANTHROPIC_API_KEY', 'claude_code_version': '2.0.42', 'output_style': 'default', 'agents': ['general-purpose', 'statusline-setup', 'Explore', 'Plan'], 'skills': [], 'plugins': [], 'uuid': '6dcc3f5a-c0a5-4560-868b-a9907f3948b5'})

AssistantMessage(content=[TextBlock(text="Hi! I'm Claude, an AI assistant built by Anthropic. I'm here to help you with a wide range of tasks, from answering questions and having conversations to helping with coding, writing, analysis, and problem-solving.\n\nIn this environment, I have access to various tools that let me:\n- **Read, write, and edit files** in your codebase\n- **Search for files and code** using patterns and keywords\n- **Run commands** in the terminal\n- **Fetch information** from the web\n- **Help with git operations** like commits and pull requests\n- **Manage tasks** with todo lists for complex projects\n\nI'm designed to be helpful, harmless, and honest. I'll do my best to understand what you need and work collaboratively with you to accomplish your goals. If I'm unsure about something or need more information, I'll ask clarifying questions.\n\nWhat would you like to work on today?")], model='claude-sonnet-4-5-20250929', parent_tool_use_id=None)

ResultMessage(subtype='success', duration_ms=7275, duration_api_ms=15910, is_error=False, num_turns=1, session_id='aab399c5-ff62-4552-a728-8e496fbd67b7', total_cost_usd=0.01317185, usage={'input_tokens': 3, 'cache_creation_input_tokens': 327, 'cache_read_input_tokens': 12442, 'output_tokens': 205, 'server_tool_use': {'web_search_requests': 0, 'web_fetch_requests': 0}, 'service_tier': 'standard', 'cache_creation': {'ephemeral_1h_input_tokens': 0, 'ephemeral_5m_input_tokens': 327}}, result="Hi! I'm Claude, an AI assistant built by Anthropic. I'm here to help you with a wide range of tasks, from answering questions and having conversations to helping with coding, writing, analysis, and problem-solving.\n\nIn this environment, I have access to various tools that let me:\n- **Read, write, and edit files** in your codebase\n- **Search for files and code** using patterns and keywords\n- **Run commands** in the terminal\n- **Fetch information** from the web\n- **Help with git operations** like commits and pull requests\n- **Manage tasks** with todo lists for complex projects\n\nI'm designed to be helpful, harmless, and honest. I'll do my best to understand what you need and work collaboratively with you to accomplish your goals. If I'm unsure about something or need more information, I'll ask clarifying questions.\n\nWhat would you like to work on today?")

This simple example shows three message types streaming back:

• SystemMessage: Initializes the agent session with configuration details including the working directory, available tools, model version, and session settings.
• AssistantMessage: Contains the agent's text response wrapped in content blocks—in this case, a single TextBlock with the introduction.
• ResultMessage: Provides final metrics about the interaction including cost, token usage, and the number of turns.

This is a particularly simple interaction where the agent completes the task in a single turn with just a text response. In more complex scenarios, you'd see additional messages streaming through—tool use blocks when the agent calls tools like Read or Bash, tool result blocks showing the output of those tools, and multiple assistant messages as the agent reasons through multiple turns. Each of these messages would arrive in real time as the agent works, giving you complete visibility into its process. Now that you understand how messages stream back, let's learn how to extract the actual text content from them.

To get the actual text the agent is saying, you need to filter the stream for AssistantMessage objects and extract text from their content blocks. Here's the complete pattern with comments explaining each step:

from claude_agent_sdk import query, AssistantMessage, TextBlock

async def main():
    async for message in query(prompt="Hi Claude! Please introduce yourself."):
        # Check if this message is an assistant response
        if isinstance(message, AssistantMessage):
            # Iterate through the content blocks in the message
            for block in message.content:
                # Check if this block contains text (vs tool uses or other content)
                if isinstance(block, TextBlock):
                    # Extract and print the actual text string
                    print(block.text)

The pattern uses isinstance() to identify AssistantMessage objects in the stream, then iterates through the content array to find TextBlock instances. This type checking is necessary because the stream can contain other message types, and even within an AssistantMessage, the content array might include tool use blocks or other types of content alongside text. The block.text attribute gives you the agent's natural language response as a string you can display, log, or process further.

When you run this code, you'll see the agent's introduction:

Hi! I'm Claude, an AI assistant built by Anthropic. I'm here to help you with a wide range of tasks, from answering questions and having conversations to helping with coding, writing, analysis, and problem-solving.

In this environment, I have access to various tools that let me:
- **Read, write, and edit files** in your codebase
- **Search for files and code** using patterns and keywords
- **Run commands** in the terminal
- **Fetch information** from the web
- **Help with git operations** like commits and pull requests
- **Manage tasks** with todo lists for complex projects

I'm designed to be helpful, harmless, and honest. I'll do my best to understand what you need and work collaboratively with you to accomplish your goals. If I'm unsure about something or need more information, I'll ask clarifying questions.

What would you like to work on today?

After the agent completes its work, the final ResultMessage provides valuable metrics about the interaction. Here's how to extract and display this information:

from claude_agent_sdk import query, ResultMessage

async def main():
    async for message in query(prompt="Hi Claude! Please introduce yourself."):
        # Handle the final result with metrics
        if isinstance(message, ResultMessage):
            print(f"\n--- Result ---")
            # The final result text (same as last AssistantMessage)
            print(f"Result: {message.result}")
            # Number of reasoning cycles the agent went through
            print(f"Turns: {message.num_turns}")
            # Total cost in USD for this interaction
            print(f"Cost: ${message.total_cost_usd:.4f}")
            # Token usage details
            print(f"Input tokens: {message.usage.get('input_tokens', 0)}")
            print(f"Output tokens: {message.usage.get('output_tokens', 0)}")

The ResultMessage includes the final output in result, the number of reasoning cycles in num_turns, the total cost in total_cost_usd, and detailed token counts in the usage dictionary. These metrics help you understand the efficiency and cost of your agent interactions—essential information when building applications that make many agent calls. Notice we use .get() when accessing the usage dictionary to safely handle cases where certain keys might not be present, providing a default value of 0.

When you run this code, you'll see the metrics output:

--- Result ---
Result: Hi! I'm Claude, an AI assistant built by Anthropic. I'm here to help you with a wide range of tasks, from answering questions and having conversations to helping with coding, writing, analysis, and problem-solving.

In this environment, I have access to various tools that let me:
- **Read, write, and edit files** in your codebase
- **Search for files and code** using patterns and keywords
- **Run commands** in the terminal
- **Fetch information** from the web
- **Help with git operations** like commits and pull requests
- **Manage tasks** with todo lists for complex projects

I'm designed to be helpful, harmless, and honest. I'll do my best to understand what you need and work collaboratively with you to accomplish your goals. If I'm unsure about something or need more information, I'll ask clarifying questions.

What would you like to work on today?
Turns: 1
Cost: $0.0132
Input tokens: 3
Output tokens: 205

This simple introduction task was completed in just one turn, used only 3 input tokens (the minimal prompt), generated 205 output tokens (the introduction text), and cost about 1.3 cents. These metrics matter because they help you understand the efficiency and cost of your agent interactions. If you're building an application that makes many agent calls, monitoring costs becomes crucial. Token counts help you optimize prompts and understand context usage. Turn counts reveal how complex the agent found the task — a simple question might complete in one turn, while a complex coding task might require multiple reasoning cycles.

You've now learned the fundamental pattern for working with the Claude Agent SDK: import the necessary types, define an async function, call query() with your prompt, iterate over streaming messages with async for, check message types with isinstance(), extract text from TextBlock objects within AssistantMessage content, and read metrics from the final ResultMessage. In the practice exercises ahead, you'll write this pattern yourself and build the hands-on experience that will serve you throughout the rest of this course.