Message Types

from isa_agent_sdk import query
 
async for msg in query("Explain async programming"):
    print(f"Type: {msg.type}")
    print(f"Content: {msg.content}")
    print(f"Timestamp: {msg.timestamp}")
    print(f"Session: {msg.session_id}")
    print(f"Metadata: {msg.metadata}")

async for msg in query("Write a function"):
    if msg.is_text:
        # Text content from the agent
        print(msg.content, end="")
 
    elif msg.is_thinking:
        # Internal reasoning (chain-of-thought)
        print(f"[Thinking: {msg.content}]")
 
    elif msg.is_tool_use:
        # Agent is calling a tool
        print(f"[Tool: {msg.tool_name}]")
 
    elif msg.is_tool_result:
        # Tool execution completed
        print(f"[Result: {msg.tool_result_value}]")
 
    elif msg.is_checkpoint:
        # Requires human input
        await msg.respond({"continue": True})
 
    elif msg.is_error:
        # Error occurred
        print(f"Error: {msg.content}")

if msg.is_tool_use:
    print(f"Tool: {msg.tool_name}")
    print(f"Arguments: {msg.tool_args}")
    print(f"Tool ID: {msg.metadata.get('tool_use_id')}")

if msg.is_tool_result:
    print(f"Tool: {msg.tool_name}")
    print(f"Result: {msg.tool_result_value}")
 
    if msg.tool_error:
        print(f"Error: {msg.tool_error}")

if msg.type == "progress":
    print(f"Progress: {msg.progress_percent}%")
    print(f"Step: {msg.progress_step}")

if msg.is_checkpoint or msg.is_hil_request:
    question = msg.metadata.get("question")
    options = msg.metadata.get("options")
    hil_type = msg.metadata.get("hil_type")
 
    print(f"Question: {question}")
 
    # Respond to continue execution
    await msg.respond({
        "authorized": True,
        "input": "user provided data"
    })

from isa_agent_sdk import AgentMessage
 
msg = AgentMessage.text("Hello, world!", session_id="session-123")

msg = AgentMessage.thinking("Let me analyze this...", session_id="session-123")

msg = AgentMessage.tool_use(
    tool_name="web_search",
    args={"query": "Python tutorials"},
    session_id="session-123",
    tool_use_id="tool-456"
)

msg = AgentMessage.tool_result(
    tool_name="web_search",
    result={"urls": ["https://..."]},
    error=None,
    session_id="session-123",
    tool_use_id="tool-456"
)

msg = AgentMessage.error(
    "Connection timeout",
    error_type="network",
    session_id="session-123"
)

msg = AgentMessage.progress(
    step="Processing file 3 of 10",
    percent=30.0,
    session_id="session-123"
)

msg = AgentMessage.hil_request(
    question="Delete these files?",
    request_type="authorization",
    options=["Yes", "No"],
    session_id="session-123"
)

from isa_agent_sdk import ConversationHistory
 
history = ConversationHistory(session_id="session-123")
 
# Add messages
history.add_user_message("Write a function")
history.add_assistant_message("Here's a function...")
 
# Access messages
print(f"Total messages: {len(history.messages)}")
print(f"Last message: {history.last_message}")
 
# Get specific content
print(f"All text: {history.get_text_content()}")
print(f"Tool calls: {history.get_tool_calls()}")
print(f"Thinking: {history.get_thinking()}")
 
# Check completion
print(f"Is complete: {history.is_complete}")

from langchain_core.messages import AIMessage, ToolMessage
from isa_agent_sdk import AgentMessage
 
# From AIMessage
ai_msg = AIMessage(content="Hello!")
agent_msg = AgentMessage.from_langchain_message(ai_msg)
 
# From ToolMessage
tool_msg = ToolMessage(content="Result", name="search")
agent_msg = AgentMessage.from_langchain_message(tool_msg)

# Get underlying isA EventData
event_data = msg.to_event_data()
print(f"Event type: {event_data.type}")
 
# Get isA EventType enum
event_type = msg.event_type
print(f"Event type enum: {event_type}")

from isa_agent_sdk._messages import TextBlock
 
block = TextBlock(text="Hello, world!")

from isa_agent_sdk._messages import ToolUseBlock
 
block = ToolUseBlock(
    id="tool-123",
    name="web_search",
    input={"query": "test"}
)

from isa_agent_sdk._messages import ToolResultBlock
 
block = ToolResultBlock(
    tool_use_id="tool-123",
    content="Search results...",
    is_error=False
)

from isa_agent_sdk._messages import AssistantMessage
 
# Create from AgentMessage
assistant_msg = AssistantMessage.from_agent_message(agent_msg)
print(f"Content blocks: {assistant_msg.content}")

from isa_agent_sdk._messages import ResultMessage
 
result = ResultMessage(
    subtype="success",
    duration_ms=1500,
    num_turns=3,
    total_cost_usd=0.02,
    result="Task completed"
)

if msg.is_streaming:
    # Streaming content (text or thinking)
    print(msg.content, end="", flush=True)
else:
    # Complete message
    print(msg.content)

if msg.type in ("node_enter", "node_exit"):
    print(f"Node: {msg.node_name}")

if msg.skill_name:
    print(f"Skill: {msg.skill_name}")

if msg.intent:
    print(f"Intent: {msg.intent}")
 
if msg.is_simple_intent:
    print("Simple query - no complex processing needed")