Skip to main content

​
Blocking Response

Use invoke() for blocking responses: 
import asyncio
from upsonic import Agent, Task, Chat


async def main():
    agent = Agent("anthropic/claude-sonnet-4-5")
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    response = await chat.invoke("What is 2+2?")
    print(response)

    task = Task(description="Explain quantum computing")
    response = await chat.invoke(task)
    print(response)


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

​
Streaming Response

Use stream() for real-time responses: 
import asyncio
from upsonic import Agent, Chat


async def main():
    agent = Agent("anthropic/claude-sonnet-4-5")
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    async for chunk in chat.stream("Tell me a story"):
        print(chunk, end="", flush=True)
    print()


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

​
Event Streaming

For full visibility into tool calls and execution, use events=True to stream events instead of text: 
import asyncio
from upsonic import Agent, Chat
from upsonic.run.events.events import TextDeltaEvent, ToolCallDeltaEvent, ToolResultEvent
from upsonic.tools import tool


@tool
def calculate(x: int, y: int) -> int:
    """Add two numbers."""
    return x + y


async def main():
    agent = Agent("anthropic/claude-sonnet-4-5", tools=[calculate])
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    print("Assistant: ", end="", flush=True)

    async for event in chat.stream("Calculate 5 + 3", events=True):
        if isinstance(event, ToolCallDeltaEvent):
            # Tool call streaming (name and arguments)
            if event.tool_name:
                print(f"\n[Tool: {event.tool_name}] ", end="", flush=True)
            if event.args_delta:
                print(event.args_delta, end="", flush=True)

        elif isinstance(event, ToolResultEvent):
            # Tool execution result
            print(f"\n[Result: {event.result}]")
            print("Assistant: ", end="", flush=True)

        elif isinstance(event, TextDeltaEvent):
            # LLM text response streaming
            print(event.content, end="", flush=True)

    print()


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

​
Using with invoke()

You can also use events with invoke(): 
import asyncio
from upsonic import Agent, Chat
from upsonic.run.events.events import TextDeltaEvent, ToolCallDeltaEvent, ToolResultEvent
from upsonic.tools import tool


@tool
def calculate(x: int, y: int) -> int:
    """Add two numbers."""
    return x + y


async def main():
    agent = Agent("anthropic/claude-sonnet-4-5", tools=[calculate])
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    print("Assistant: ", end="", flush=True)

    async for event in await chat.invoke("Calculate 5 + 3", stream=True, events=True):
        if isinstance(event, ToolCallDeltaEvent):
            # Tool call streaming (name and arguments)
            if event.tool_name:
                print(f"\n[Tool: {event.tool_name}] ", end="", flush=True)
            if event.args_delta:
                print(event.args_delta, end="", flush=True)

        elif isinstance(event, ToolResultEvent):
            # Tool execution result
            print(f"\n[Result: {event.result}]")
            print("Assistant: ", end="", flush=True)

        elif isinstance(event, TextDeltaEvent):
            # LLM text response streaming
            print(event.content, end="", flush=True)

    print()


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

​
With Attachments

Pass file paths via context parameter: 
import asyncio
from upsonic import Agent, Chat


async def main():
    agent = Agent("anthropic/claude-sonnet-4-5")
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    response = await chat.invoke(
        "Analyze this document",
        context=["document.pdf", "data.csv"]
    )
    print(response)


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

​
Accessing History

Get messages as ChatMessage objects: 
import asyncio
from upsonic import Agent, Chat


async def main():
    agent = Agent("anthropic/claude-sonnet-4-5")
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    await chat.invoke("Hello")
    await chat.invoke("How are you?")

    for msg in chat.all_messages:
        print(f"{msg.role}: {msg.content}")

    recent = chat.get_recent_messages(count=5)
    print(f"Recent: {len(recent)} messages")


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

​
Session State

Check and manage session state: 
import asyncio
from upsonic import Agent, Chat


async def main():
    agent = Agent("anthropic/claude-sonnet-4-5")
    chat = Chat(session_id="session1", user_id="user1", agent=agent)

    await chat.invoke("Hello")

    print(f"State: {chat.state.value}")
    print(f"Duration: {chat.duration:.1f}s")


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