Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -31,14 +31,12 @@ anyio.run(main)
`query()` is an async function for querying Claude Code. It returns an `AsyncIterator` of response messages. See [src/claude_agent_sdk/query.py](src/claude_agent_sdk/query.py).

```python
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage

# 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)
print(message)

# With options
options = ClaudeAgentOptions(
Expand Down
12 changes: 3 additions & 9 deletions examples/quick_start.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,7 @@ async def basic_example():

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}")
print(f"Claude: {message}")
print()


Expand All @@ -37,9 +35,7 @@ async def with_options_example():
prompt="Explain what Python is in one sentence.", options=options
):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
print(f"Claude: {message}")
print()


Expand All @@ -57,9 +53,7 @@ async def with_tools_example():
options=options,
):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
print(f"Claude: {message}")
elif isinstance(message, ResultMessage) and message.total_cost_usd > 0:
print(f"\nCost: ${message.total_cost_usd:.4f}")
print()
Expand Down
8 changes: 2 additions & 6 deletions examples/streaming_mode.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,13 +42,9 @@ def display_message(msg):
- ResultMessage: "Result ended" + cost if available
"""
if isinstance(msg, UserMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"User: {block.text}")
print(f"User: {msg}")
elif isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
print(f"Claude: {msg}")
elif isinstance(msg, SystemMessage):
# Ignore system messages
pass
Expand Down
28 changes: 7 additions & 21 deletions examples/streaming_mode_ipython.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,7 @@

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}")
print(f"Claude: {msg}")


# ============================================================================
Expand All @@ -41,9 +39,7 @@ async def send_and_receive(prompt):
await client.query(prompt)
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}")
print(f"Claude: {msg}")

await send_and_receive("Tell me a short joke")
print("\n---\n")
Expand All @@ -65,9 +61,7 @@ async def send_and_receive(prompt):
async def get_response():
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}")
print(f"Claude: {msg}")


# Use it multiple times
Expand Down Expand Up @@ -106,9 +100,7 @@ async def consume_messages():
async for msg in client.receive_messages():
messages_received.append(msg)
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
print(f"Claude: {msg}")

# Check if we got a result after interrupt
if isinstance(msg, ResultMessage) and interrupt_sent:
Expand Down Expand Up @@ -154,9 +146,7 @@ async def consume_messages():
async for msg in client.receive_response():
messages.append(msg)
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
print(f"Claude: {msg}")

except asyncio.TimeoutError:
print("Request timed out after 20 seconds")
Expand Down Expand Up @@ -201,9 +191,7 @@ async def message_generator():

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}")
print(f"Claude: {msg}")


# ============================================================================
Expand All @@ -222,8 +210,6 @@ async def message_generator():
# Process them afterwards
for msg in messages:
if isinstance(msg, AssistantMessage):
for block in msg.content:
if isinstance(block, TextBlock):
print(f"Claude: {block.text}")
print(f"Claude: {msg}")
elif isinstance(msg, ResultMessage):
print(f"Total messages: {len(messages)}")
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "hatchling.build"

[project]
name = "claude-agent-sdk"
version = "0.1.6"
version = "0.1.7"
description = "Python SDK for Claude Code"
readme = "README.md"
requires-python = ">=3.10"
Expand Down
2 changes: 1 addition & 1 deletion src/claude_agent_sdk/_version.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
"""Version information for claude-agent-sdk."""

__version__ = "0.1.6"
__version__ = "0.1.7"
150 changes: 150 additions & 0 deletions src/claude_agent_sdk/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,17 @@
if TYPE_CHECKING:
from mcp.server import Server as McpServer

# Repr/str truncation length for content blocks
_REPR_MAX_LENGTH = 50


def _truncate_repr(text: str) -> str:
"""Truncate text for repr output if it exceeds _REPR_MAX_LENGTH."""
if len(text) > _REPR_MAX_LENGTH:
return text[:_REPR_MAX_LENGTH] + "..."
return text


# Permission modes
PermissionMode = Literal["default", "acceptEdits", "plan", "bypassPermissions"]

Expand Down Expand Up @@ -127,6 +138,18 @@ class PermissionResultAllow:
updated_input: dict[str, Any] | None = None
updated_permissions: list[PermissionUpdate] | None = None

def __repr__(self) -> str:
"""Return technical representation showing structure."""
return (
f"PermissionResultAllow(behavior='allow', "
f"updated_input={self.updated_input is not None}, "
f"updated_permissions={len(self.updated_permissions) if self.updated_permissions else 0})"
)

def __str__(self) -> str:
"""Return user-friendly permission result info."""
return "PermissionResultAllow"


@dataclass
class PermissionResultDeny:
Expand All @@ -136,6 +159,18 @@ class PermissionResultDeny:
message: str = ""
interrupt: bool = False

def __repr__(self) -> str:
"""Return technical representation showing structure."""
message_preview = _truncate_repr(self.message)
return (
f"PermissionResultDeny(behavior='deny', message={message_preview!r}, "
f"interrupt={self.interrupt})"
)

def __str__(self) -> str:
"""Return user-friendly permission result info."""
return f"PermissionResultDeny: {self.message}"


PermissionResult = PermissionResultAllow | PermissionResultDeny

Expand Down Expand Up @@ -423,6 +458,15 @@ class TextBlock:

text: str

def __repr__(self) -> str:
"""Return technical representation showing structure."""
preview = _truncate_repr(self.text)
return f"TextBlock({preview!r})"

def __str__(self) -> str:
"""Return user-friendly text content."""
return self.text


@dataclass
class ThinkingBlock:
Expand All @@ -431,6 +475,15 @@ class ThinkingBlock:
thinking: str
signature: str

def __repr__(self) -> str:
"""Return technical representation showing structure."""
thinking_preview = _truncate_repr(self.thinking)
return f"ThinkingBlock(thinking={thinking_preview!r}, signature={self.signature!r})"

def __str__(self) -> str:
"""Return user-friendly thinking content."""
return self.thinking


@dataclass
class ToolUseBlock:
Expand All @@ -440,6 +493,17 @@ class ToolUseBlock:
name: str
input: dict[str, Any]

def __repr__(self) -> str:
"""Return technical representation showing structure."""
input_preview = _truncate_repr(str(self.input))
return (
f"ToolUseBlock(id={self.id!r}, name={self.name!r}, input={input_preview})"
)

def __str__(self) -> str:
"""Return user-friendly tool usage info."""
return f"Tool: {self.name} (ID: {self.id})"


@dataclass
class ToolResultBlock:
Expand All @@ -449,6 +513,22 @@ class ToolResultBlock:
content: str | list[dict[str, Any]] | None = None
is_error: bool | None = None

def __repr__(self) -> str:
"""Return technical representation showing structure."""
if isinstance(self.content, str):
content_preview = _truncate_repr(self.content)
else:
content_preview = _truncate_repr(str(self.content))
return (
f"ToolResultBlock(tool_use_id={self.tool_use_id!r}, "
f"content={content_preview!r}, is_error={self.is_error})"
)

def __str__(self) -> str:
"""Return user-friendly tool result info."""
status = "error" if self.is_error else "success"
return f"Tool Result [{status}] (ID: {self.tool_use_id})"


ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock

Expand All @@ -461,6 +541,25 @@ class UserMessage:
content: str | list[ContentBlock]
parent_tool_use_id: str | None = None

def __repr__(self) -> str:
"""Return technical representation showing structure."""
if isinstance(self.content, str):
content_preview = _truncate_repr(self.content)
else:
# Show repr of each content block
blocks_repr = ", ".join(repr(b) for b in self.content)
content_preview = f"[{blocks_repr}]"
return f"UserMessage(content={content_preview!r})"

def __str__(self) -> str:
"""Return user-friendly message content."""
if isinstance(self.content, str):
return self.content
else:
# Use str() of each content block for user-friendly output
block_strs = "\n ".join(str(b) for b in self.content)
return f"UserMessage:\n {block_strs}"


@dataclass
class AssistantMessage:
Expand All @@ -470,6 +569,19 @@ class AssistantMessage:
model: str
parent_tool_use_id: str | None = None

def __repr__(self) -> str:
"""Return technical representation showing structure."""
# Show repr of each content block
blocks_repr = ", ".join(repr(b) for b in self.content)
content_preview = f"[{blocks_repr}]"
return f"AssistantMessage(model={self.model!r}, content={content_preview})"

def __str__(self) -> str:
"""Return user-friendly message summary."""
# Use str() of each content block for user-friendly output
block_strs = "\n ".join(str(b) for b in self.content)
return f"AssistantMessage from {self.model}:\n {block_strs}"


@dataclass
class SystemMessage:
Expand All @@ -478,6 +590,15 @@ class SystemMessage:
subtype: str
data: dict[str, Any]

def __repr__(self) -> str:
"""Return technical representation showing structure."""
data_preview = _truncate_repr(str(self.data))
return f"SystemMessage(subtype={self.subtype!r}, data={data_preview})"

def __str__(self) -> str:
"""Return user-friendly system message info."""
return f"SystemMessage [{self.subtype}]"


@dataclass
class ResultMessage:
Expand All @@ -493,6 +614,22 @@ class ResultMessage:
usage: dict[str, Any] | None = None
result: str | None = None

def __repr__(self) -> str:
"""Return technical representation showing structure."""
return (
f"ResultMessage(subtype={self.subtype!r}, duration_ms={self.duration_ms}, "
f"is_error={self.is_error}, num_turns={self.num_turns}, "
f"session_id={self.session_id!r})"
)

def __str__(self) -> str:
"""Return user-friendly result message info."""
status = "error" if self.is_error else "success"
return (
f"ResultMessage [{status}] - {self.num_turns} turn(s), "
f"{self.duration_ms}ms (API: {self.duration_api_ms}ms)"
)


@dataclass
class StreamEvent:
Expand All @@ -503,6 +640,19 @@ class StreamEvent:
event: dict[str, Any] # The raw Anthropic API stream event
parent_tool_use_id: str | None = None

def __repr__(self) -> str:
"""Return technical representation showing structure."""
event_preview = _truncate_repr(str(self.event))
return (
f"StreamEvent(uuid={self.uuid!r}, session_id={self.session_id!r}, "
f"event={event_preview})"
)

def __str__(self) -> str:
"""Return user-friendly stream event info."""
event_type = self.event.get("type", "unknown")
return f"StreamEvent [{event_type}] ({self.uuid[:8]}...)"


Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

Expand Down
Loading