Usage Guide

Learn how to build AI agents with the Rowboat Python SDK

Initialize the Client

Start by creating a Client instance with your credentials:

from rowboat.client import Client

client = Client(
    host="https://api.rowboat.dev",
    projectId="your-project-id",
    apiKey="your-api-key"
)

Basic Conversation

Starting a New Conversation

Create a new conversation by calling run_turn() with a list of messages:

from rowboat.schema import UserMessage

result = client.run_turn(
    messages=[
        UserMessage(role='user', content="list my github repos")
    ]
)

# Access the assistant's response
print(result.turn.output[-1].content)

# Save the conversation ID for future turns
conversation_id = result.conversationId
print("Conversation ID:", conversation_id)

Continuing a Conversation

To continue a conversation, pass the conversationId from the previous turn:

result = client.run_turn(
    messages=[
        UserMessage(role='user', content="how many did you find?")
    ],
    conversationId=conversation_id
)

print(result.turn.output[-1].content)

The SDK is stateless - you must manage conversation IDs to maintain context across turns.

Message Types

Rowboat supports multiple message types for different conversation participants:

User Messages

Messages from end users:

from rowboat.schema import UserMessage

message = UserMessage(
    role='user',
    content="What's the weather in San Francisco?"
)

System Messages

Provide system instructions or context:

from rowboat.schema import SystemMessage

message = SystemMessage(
    role='system',
    content="You are a helpful weather assistant."
)

Assistant Messages

Agent responses with optional metadata:

from rowboat.schema import AssistantMessage

message = AssistantMessage(
    role='assistant',
    content="The weather in San Francisco is sunny and 72°F.",
    responseType='external',
    agenticName='weather-bot'
)

Tool Messages

Results from tool executions:

from rowboat.schema import ToolMessage

message = ToolMessage(
    role='tool',
    content='{"temperature": 72, "condition": "sunny"}',
    toolCallId='call_123',
    toolName='weather_lookup'
)

Using Mock Tools

Test your agent’s behavior without executing real tools by providing mock responses:

result = client.run_turn(
    messages=[
        UserMessage(role='user', content="What's the weather?")
    ],
    mockTools={
        "weather_lookup": "The weather in any city is sunny and 25°C.",
        "calculator": "The result of any calculation is 42."
    }
)

print(result.turn.output[-1].content)

Mock tools are perfect for:

• Testing agent logic without external dependencies
• Simulating specific tool responses
• Development and debugging

Multi-Turn Conversations

Here’s a complete example of a multi-turn conversation:

from rowboat.client import Client
from rowboat.schema import UserMessage

client = Client(
    host="https://api.rowboat.dev",
    projectId="your-project-id",
    apiKey="your-api-key"
)

# Turn 1: Start conversation
result = client.run_turn(
    messages=[
        UserMessage(role='user', content="Find my most recent pull request")
    ]
)
conversation_id = result.conversationId
print("Turn 1:", result.turn.output[-1].content)

# Turn 2: Follow-up question
result = client.run_turn(
    messages=[
        UserMessage(role='user', content="What's the status of that PR?")
    ],
    conversationId=conversation_id
)
print("Turn 2:", result.turn.output[-1].content)

# Turn 3: Another follow-up
result = client.run_turn(
    messages=[
        UserMessage(role='user', content="Can you merge it?")
    ],
    conversationId=conversation_id
)
print("Turn 3:", result.turn.output[-1].content)

Conversation Management

Storing Conversation IDs

Store conversation IDs in your database to enable persistent conversations:

import sqlite3

def save_conversation(user_id: str, conversation_id: str):
    conn = sqlite3.connect('conversations.db')
    cursor = conn.cursor()
    cursor.execute(
        'INSERT INTO conversations (user_id, conversation_id) VALUES (?, ?)',
        (user_id, conversation_id)
    )
    conn.commit()
    conn.close()

def get_conversation(user_id: str) -> str:
    conn = sqlite3.connect('conversations.db')
    cursor = conn.cursor()
    cursor.execute(
        'SELECT conversation_id FROM conversations WHERE user_id = ? ORDER BY created_at DESC LIMIT 1',
        (user_id,)
    )
    result = cursor.fetchone()
    conn.close()
    return result[0] if result else None

Creating New Conversations

Create a new conversation by omitting the conversationId:

# New conversation - no conversationId
result = client.run_turn(
    messages=[UserMessage(role='user', content="Hello!")]
)
new_conversation_id = result.conversationId

Error Handling

Handle API errors gracefully:

from rowboat.client import Client
from rowboat.schema import UserMessage

client = Client(
    host="https://api.rowboat.dev",
    projectId="your-project-id",
    apiKey="your-api-key"
)

try:
    result = client.run_turn(
        messages=[
            UserMessage(role='user', content="Hello!")
        ]
    )
    print(result.turn.output[-1].content)
except ValueError as e:
    print(f"API Error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Complete Example

Here’s a complete example bringing it all together:

import os
from rowboat.client import Client
from rowboat.schema import UserMessage

def main():
    # Initialize client
    client = Client(
        host=os.getenv("ROWBOAT_HOST", "https://api.rowboat.dev"),
        projectId=os.getenv("ROWBOAT_PROJECT_ID"),
        apiKey=os.getenv("ROWBOAT_API_KEY")
    )
    
    conversation_id = None
    
    print("Rowboat Chat (type 'quit' to exit)")
    print("-" * 40)
    
    while True:
        user_input = input("You: ").strip()
        
        if user_input.lower() == 'quit':
            break
            
        if not user_input:
            continue
        
        try:
            # Send message to agent
            result = client.run_turn(
                messages=[
                    UserMessage(role='user', content=user_input)
                ],
                conversationId=conversation_id
            )
            
            # Update conversation ID
            conversation_id = result.conversationId
            
            # Print agent response
            response = result.turn.output[-1].content
            print(f"Agent: {response}\n")
            
        except ValueError as e:
            print(f"Error: {e}\n")
        except KeyboardInterrupt:
            break

if __name__ == "__main__":
    main()

Best Practices

Manage Conversation State

Store conversation IDs in a database or cache to enable persistent conversations across sessions.

Handle Errors Gracefully

Always wrap API calls in try-except blocks to handle network errors and API failures.

Secure Your API Keys

Never hardcode API keys. Use environment variables or secure key management systems.

Use Mock Tools for Testing

Test your agent logic using mock tools before deploying with real tool integrations.

Implement Rate Limiting

Add rate limiting to prevent excessive API calls and protect your quota.

Next Steps

API Reference

Explore the complete API documentation