Type System

Provider-agnostic types that abstract away SDK-specific implementations. Use these types throughout your code to support multiple AI providers.

from shared import Message, Part, Role, ToolSchema, ...
Provider Independence
These types are defined in shared/plugins/model_provider/types.py and work identically regardless of which AI provider you're using.
Import all types
from jaato import (
    # Enums
    Role,
    FinishReason,
    ClientType,

    # Message types
    Message,
    Part,

    # Tool types
    ToolSchema,
    FunctionCall,
    ToolResult,

    # Response types
    ProviderResponse,
    TokenUsage,

    # Display context
    PresentationContext,

    # Attachments
    Attachment,
)

Enums

Role

Identifies the sender of a message in the conversation.

Value Description
Role.USER Message from the user
Role.MODEL Message from the AI model
Role.TOOL Tool execution result

FinishReason

Indicates why the model stopped generating.

Value Description
FinishReason.STOP Natural completion
FinishReason.MAX_TOKENS Hit token limit
FinishReason.TOOL_USE Model wants to use a tool
FinishReason.SAFETY Safety filter triggered
FinishReason.ERROR Error occurred
FinishReason.CANCELLED Cancelled via CancelToken
FinishReason.UNKNOWN Unknown reason

ClientType

Presentation-layer category for PresentationContext. Describes the kind of display surface, not specific apps — a Telegram bot and a Slack bot are both CHAT.

Value Description
ClientType.TERMINAL TUI / CLI (rich text, fixed-width)
ClientType.WEB Browser-based UI (HTML, responsive)
ClientType.CHAT Messaging platform (Telegram, Slack, WhatsApp, …)
ClientType.API Headless / programmatic (plain text)
String Enum
ClientType inherits from str, so ClientType.CHAT == "chat" is True. Invalid values raise ValueError.
Using enums
from jaato import Role, FinishReason, ClientType, Message

# Create message with role
msg = Message.from_text(Role.USER, "Hello!")

# Check role
if msg.role == Role.MODEL:
    print("From model")

# Check finish reason
response = client.send_message(...)
if response.finish_reason == FinishReason.TOOL_USE:
    print("Model wants to use a tool")
elif response.finish_reason == FinishReason.STOP:
    print("Natural completion")

Message

Represents a single message in the conversation. Contains a role and one or more parts.

Properties

  • role Role
    Who sent the message
  • parts List[Part]
    Content parts of the message
  • text Optional[str] property
    Extracted text content (convenience)
  • function_calls List[FunctionCall] property
    Extracted function calls (convenience)

Class Methods

Message.from_text(role: Role, text: str) -> Message

Creates a simple text message.

Working with messages
from jaato import Message, Part, Role

# Create a text message
msg = Message.from_text(Role.USER, "Hello!")

# Create with multiple parts
msg = Message(
    role=Role.USER,
    parts=[
        Part.from_text("What's in this image?"),
        Part(inline_data=Attachment(...))
    ]
)

# Access properties
print(f"Role: {msg.role}")
print(f"Text: {msg.text}")

# Get function calls from model message
for call in msg.function_calls:
    print(f"Call: {call.name}({call.args})")
Working with history
# Get conversation history
history = client.get_history()

for msg in history:
    if msg.role == Role.USER:
        print(f"User: {msg.text}")
    elif msg.role == Role.MODEL:
        if msg.function_calls:
            for call in msg.function_calls:
                print(f"Tool: {call.name}")
        else:
            print(f"Model: {msg.text}")

Part

A single piece of content within a message. Can be text, a function call, a function response, or inline data (images, etc.).

Properties

  • text Optional[str]
    Text content
  • function_call Optional[FunctionCall]
    Function call from model
  • function_response Optional[ToolResult]
    Tool execution result
  • inline_data Optional[Attachment]
    Binary data (images, etc.)

Class Methods

Part.from_text(text: str) -> Part
Part.from_function_call(call: FunctionCall) -> Part
Part.from_function_response(result: ToolResult) -> Part
Creating parts
from jaato import Part, FunctionCall, ToolResult

# Text part
text_part = Part.from_text("Hello world")

# Function call part
call = FunctionCall(
    id="call_123",
    name="get_weather",
    args={"city": "NYC"}
)
call_part = Part.from_function_call(call)

# Function response part
result = ToolResult(
    call_id="call_123",
    name="get_weather",
    result="Sunny, 72F"
)
response_part = Part.from_function_response(result)

# Inline data (image)
image_part = Part(
    inline_data=Attachment(
        mime_type="image/png",
        data=image_bytes
    )
)
Checking part type
for part in message.parts:
    if part.text:
        print(f"Text: {part.text}")
    elif part.function_call:
        print(f"Call: {part.function_call.name}")
    elif part.function_response:
        print(f"Result: {part.function_response.result}")
    elif part.inline_data:
        print(f"Data: {part.inline_data.mime_type}")

ToolSchema

Declares a tool that the model can invoke. Used by plugins to describe their available tools.

Properties

  • name str required
    Unique tool name (used for invocation)
  • description str required
    Human-readable description for the model
  • parameters Dict required
    JSON Schema for parameters
JSON Schema
The parameters field follows JSON Schema format with type, properties, and required fields.
Defining tool schemas
from jaato import ToolSchema

# Simple tool
weather_tool = ToolSchema(
    name="get_weather",
    description="Get current weather for a city",
    parameters={
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "City name"
            },
            "units": {
                "type": "string",
                "enum": ["celsius", "fahrenheit"],
                "description": "Temperature units"
            }
        },
        "required": ["city"]
    }
)

# Tool with complex parameters
search_tool = ToolSchema(
    name="search_files",
    description="Search for files matching criteria",
    parameters={
        "type": "object",
        "properties": {
            "pattern": {
                "type": "string",
                "description": "Glob pattern"
            },
            "max_results": {
                "type": "integer",
                "description": "Maximum results",
                "default": 10
            },
            "include_hidden": {
                "type": "boolean",
                "default": False
            }
        },
        "required": ["pattern"]
    }
)

FunctionCall

Represents a tool invocation requested by the model. Contains the tool name and arguments.

Properties

  • id str required
    Unique identifier for this call
  • name str required
    Name of the tool to invoke
  • args Dict[str, Any] required
    Arguments for the tool
Working with function calls
from jaato import FunctionCall

# Create a function call
call = FunctionCall(
    id="call_abc123",
    name="get_weather",
    args={"city": "New York", "units": "fahrenheit"}
)

# Access properties
print(f"ID: {call.id}")
print(f"Name: {call.name}")
print(f"Args: {call.args}")

# Extract from model response
response = client.send_message(...)
if response.function_calls:
    for call in response.function_calls:
        print(f"Model wants to call: {call.name}")
        print(f"With args: {call.args}")

ToolResult

Contains the result of executing a tool. Sent back to the model after tool execution.

Properties

  • call_id str required
    ID of the function call this responds to
  • name str required
    Name of the tool that was executed
  • result Any required
    The tool's output (usually string or dict)
  • is_error bool optional
    Whether this is an error result
    Default: False
  • attachments List[Attachment] optional
    Binary attachments (images, etc.)
Creating tool results
from jaato import ToolResult, Attachment

# Successful result
result = ToolResult(
    call_id="call_abc123",
    name="get_weather",
    result="Sunny, 72°F in New York"
)

# Error result
error_result = ToolResult(
    call_id="call_xyz789",
    name="read_file",
    result="File not found: /path/to/file",
    is_error=True
)

# Result with attachment
screenshot_result = ToolResult(
    call_id="call_screenshot",
    name="take_screenshot",
    result="Screenshot captured",
    attachments=[
        Attachment(
            mime_type="image/png",
            data=screenshot_bytes,
            display_name="screenshot.png"
        )
    ]
)

ProviderResponse

Unified response from any AI provider. Contains the response content, usage stats, and finish reason.

Properties

  • text Optional[str]
    Text response from the model
  • function_calls List[FunctionCall]
    Tool calls requested by the model
  • usage Optional[TokenUsage]
    Token usage statistics
  • finish_reason FinishReason
    Why generation stopped
  • raw Any
    Raw provider-specific response object
  • structured_output Optional[Any]
    Parsed structured output (if requested)

Convenience Properties

  • has_function_calls bool property
    True if response contains function calls
  • has_structured_output bool property
    True if response has structured output

TokenUsage

  • prompt_tokens int
    Tokens in the input
  • output_tokens int
    Tokens in the output
  • total_tokens int
    Total tokens used
Working with responses
from jaato import FinishReason

# Response handling
response = provider.send_message("Hello")

# Check for text
if response.text:
    print(f"Response: {response.text}")

# Check for function calls
if response.has_function_calls:
    for call in response.function_calls:
        print(f"Tool: {call.name}")

# Check finish reason
if response.finish_reason == FinishReason.STOP:
    print("Completed normally")
elif response.finish_reason == FinishReason.MAX_TOKENS:
    print("Hit token limit")

# Token usage
if response.usage:
    print(f"Prompt: {response.usage.prompt_tokens}")
    print(f"Output: {response.usage.output_tokens}")
    print(f"Total: {response.usage.total_tokens}")

CancelToken

Thread-safe cancellation signaling for stopping long-running operations. Used to gracefully interrupt message processing, streaming, or tool execution.

Methods

  • cancel() None
    Signal cancellation. Idempotent (safe to call multiple times).
  • wait(timeout) bool
    Block until cancelled or timeout. Returns True if cancelled.
  • raise_if_cancelled() None
    Raises CancelledException if cancelled.
  • on_cancel(callback) None
    Register callback to run when cancelled.
  • reset() None
    Clear cancelled state (for reuse).

Properties

  • is_cancelled bool property
    Check if cancellation was requested.
Using CancelToken
from jaato import CancelToken, CancelledException

# Create token
token = CancelToken()

# Check state
if not token.is_cancelled:
    print("Still running")

# Signal cancellation (from another thread)
token.cancel()

# Check for cancellation
token.raise_if_cancelled()  # Raises CancelledException

# Register callback
def on_cancel():
    print("Cleanup...")
token.on_cancel(on_cancel)

# Wait for cancellation (with timeout)
was_cancelled = token.wait(timeout=5.0)
With JaatoClient
import threading

client = JaatoClient()
# ... connect and configure ...

# Cancel from another thread
def cancel_after_delay():
    time.sleep(5)
    client.stop()

threading.Thread(target=cancel_after_delay).start()

try:
    response = client.send_message(
        "Long task...",
        on_output=handler
    )
except CancelledException:
    print("Request was cancelled")

CancelledException

Exception raised when an operation is cancelled via CancelToken. Inherits from Exception.

Constructor

CancelledException(message: str = "Operation was cancelled")

When Raised

  • During streaming when CancelToken.cancel() is called
  • During retry backoff sleep
  • When raise_if_cancelled() is called on a cancelled token
  • Before each retry attempt if token is cancelled
Finish Reason
When a request is cancelled, the response will have finish_reason == FinishReason.CANCELLED.
Handling CancelledException
from jaato import CancelledException, FinishReason

try:
    response = client.send_message(
        "Generate a long document...",
        on_output=handler
    )
    if response.finish_reason == FinishReason.CANCELLED:
        print("Generation was cancelled")
except CancelledException as e:
    print(f"Cancelled: {e}")
    # Handle partial results if needed
In retry context
from shared.retry_utils import with_retry, RetryConfig
from jaato import CancelToken, CancelledException

token = CancelToken()

try:
    result, stats = with_retry(
        lambda: api_call(),
        config=RetryConfig(max_attempts=5),
        cancel_token=token
    )
except CancelledException:
    print("Retries cancelled")
    # stats.attempts shows how many were tried

PresentationContext

Display capabilities and constraints of the connected client. Assembled by each client at connection time and transmitted to the server via ClientConfigRequest.presentation. The model receives a compact instruction block derived from this context, letting it adapt output format (e.g. vertical key-value lists on narrow screens, full tables on wide ones).

Properties

  • content_width int
    Available character width for content
    Default: 80
  • content_height Optional[int]
    Available line height (if known)
  • supports_markdown bool
    Client renders markdown
    Default: True
  • supports_tables bool
    Client renders markdown tables
    Default: True
  • supports_code_blocks bool
    Client renders fenced code blocks
    Default: True
  • supports_images bool
    Client can display inline images
    Default: False
  • supports_rich_text bool
    Client renders bold, italic, etc.
    Default: True
  • supports_unicode bool
    Client can display Unicode characters
    Default: True
  • supports_mermaid bool
    Client renders Mermaid diagrams
    Default: False
  • supports_expandable_content bool
    Client has expand/collapse widgets for overflow
    Default: False
  • client_type ClientType
    Presentation surface category
    Default: ClientType.TERMINAL

Methods

to_system_instruction() -> str

Generate the compact display-context block injected into system instructions.

to_dict() -> Dict[str, Any]

Serialize for event transport. client_type is serialized as its string value.

PresentationContext.from_dict(data: Dict) -> PresentationContext

Construct from a dict (e.g. from ClientConfigRequest.presentation).

Client declarations
from jaato import PresentationContext, ClientType

# TUI terminal
ctx = PresentationContext(
    content_width=terminal_width - 4,
    client_type=ClientType.TERMINAL,
)

# Chat bot (Telegram, Slack, …)
ctx = PresentationContext(
    content_width=45,
    supports_tables=False,
    supports_images=True,
    supports_expandable_content=True,
    client_type=ClientType.CHAT,
)

# Headless API
ctx = PresentationContext(
    content_width=120,
    supports_markdown=False,
    supports_tables=False,
    supports_code_blocks=False,
    supports_images=False,
    client_type=ClientType.API,
)
Setting on client/session
# Via JaatoClient
client.set_presentation_context(ctx)

# Via JaatoSession
session.set_presentation_context(ctx)

# Serialization round-trip
d = ctx.to_dict()       # {"content_width": 45, "client_type": "chat", ...}
ctx2 = PresentationContext.from_dict(d)

Attachment

Binary data attachment for multimodal content. Used for images, files, and other non-text data.

Properties

  • mime_type str required
    MIME type (e.g., image/png)
  • data bytes required
    Raw binary data
  • display_name Optional[str] optional
    Human-readable name

Common MIME Types

Type MIME
PNG image image/png
JPEG image image/jpeg
GIF image image/gif
WebP image image/webp
PDF document application/pdf
Creating attachments
from jaato import Attachment, Part, Message, Role

# From file
with open("image.png", "rb") as f:
    image_data = f.read()

attachment = Attachment(
    mime_type="image/png",
    data=image_data,
    display_name="screenshot.png"
)

# Use in message
message = Message(
    role=Role.USER,
    parts=[
        Part.from_text("What's in this image?"),
        Part(inline_data=attachment)
    ]
)

# Send multimodal message
response = client.send_message_with_parts(
    message.parts,
    on_output=on_output
)
In tool results
# Tool that returns an image
def take_screenshot():
    screenshot = capture_screen()
    return ToolResult(
        call_id=call.id,
        name="take_screenshot",
        result="Screenshot captured",
        attachments=[
            Attachment(
                mime_type="image/png",
                data=screenshot,
                display_name="screen.png"
            )
        ]
    )