🔧

Ripperdoc

SDK Overview

Using Ripperdoc as a Python library

Ripperdoc provides a Python SDK for building AI-powered applications with programmatic control over conversations, tools, and permissions.

Installation

pip install ripperdoc-agent-sdk

Quick Start

One-Shot Query

For simple, single-turn queries:

import asyncio
from ripperdoc_agent_sdk import query, RipperdocAgentOptions

async def main():
    options = RipperdocAgentOptions(
        model="main",
        permission_mode="default"
    )

    async for message in query(
        prompt="Explain what this function does",
        options=options
    ):
        if message.type == "assistant":
            for block in message.content:
                if block.type == "text":
                    print(block.text)

asyncio.run(main())

Session-Based Client

For multi-turn conversations with full control:

import asyncio
from ripperdoc_agent_sdk import RipperdocSDKClient, RipperdocAgentOptions

async def main():
    options = RipperdocAgentOptions(
        model="main",
        permission_mode="acceptEdits",
        verbose=True
    )

    async with RipperdocSDKClient(options=options) as client:
        # Start a conversation
        await client.query("What files are in this project?")
        async for message in client.receive_messages():
            if message.type == "assistant":
                for block in message.content:
                    if block.type == "text":
                        print(block.text)

        # Continue the conversation
        await client.query("Read the main.py file")
        async for message in client.receive_messages():
            if message.type == "assistant":
                for block in message.content:
                    if block.type == "text":
                        print(block.text)

asyncio.run(main())

RipperdocAgentOptions

The RipperdocAgentOptions class configures SDK behavior:

from ripperdoc_agent_sdk import RipperdocAgentOptions

options = RipperdocAgentOptions(
    # Model configuration
    model="main",                   # Model profile name
    max_thinking_tokens=0,          # Extended thinking tokens (0 = disabled)

    # Permission mode (replaces yolo_mode)
    permission_mode="default",      # "default" | "acceptEdits" | "bypassPermissions" | "plan"

    # Tool configuration
    allowed_tools=["Read", "Glob"],  # Whitelist specific tools
    disallowed_tools=["Bash"],       # Blacklist specific tools

    # Behavior settings
    verbose=False,                  # Enable verbose output
    max_turns=None,                 # Max conversation turns (None = unlimited)
    query_timeout=300.0,            # Query timeout in seconds (default: 300)

    # Context
    cwd=None,                       # Working directory path
    context={},                     # Custom context variables
    system_prompt=None,             # Override system prompt
    additional_instructions=None,   # Additional instructions

    # Session management
    resume=None,                    # Session ID to resume from
    continue_conversation=False,    # Continue most recent conversation
    fork_session=False,             # Create new branch when resuming

    # Custom permissions
    permission_checker=None,        # Custom permission checker function

    # MCP servers
    mcp_servers=None,               # MCP server configurations

    # Programmatic agents
    agents=None,                    # Custom subagent definitions

    # Programmatic hooks
    hooks=None,                     # Event hook callbacks
)

Permission Modes

The SDK supports four permission modes:

ModeDescription
defaultPrompts for dangerous operations (bash, file edits, etc.)
acceptEditsAuto-accept file edits, prompt for other dangerous operations
bypassPermissionsAllow all operations without prompting
planPlanning mode - no execution, analysis only
# Safe mode with prompts
options = RipperdocAgentOptions(permission_mode="default")

# Auto-accept file edits
options = RipperdocAgentOptions(permission_mode="acceptEdits")

# Planning mode - no execution
options = RipperdocAgentOptions(permission_mode="plan")

# Fully automated (use with caution)
options = RipperdocAgentOptions(permission_mode="bypassPermissions")

Tool Filtering

Control which tools are available:

# Only allow read operations
options = RipperdocAgentOptions(
    allowed_tools=["Read", "Glob", "Grep", "LS"]
)

# Allow all except bash
options = RipperdocAgentOptions(
    disallowed_tools=["Bash"]
)

Custom Permission Checker

Implement custom permission logic:

from ripperdoc_agent_sdk import RipperdocAgentOptions, PermissionResultAllow, PermissionResultDeny

async def my_permission_checker(tool_name, tool_input, context):
    # Auto-approve read operations
    if tool_name in ['Read', 'Glob', 'Grep']:
        return PermissionResultAllow()

    # Deny dangerous commands
    if tool_name == 'Bash':
        command = tool_input.get('command', '')
        if 'rm -rf' in command:
            return PermissionResultDeny(message="Dangerous command blocked")

    # Ask for others
    return PermissionResultAllow()

options = RipperdocAgentOptions(
    permission_checker=my_permission_checker
)

Message Types

The SDK yields different message types with structured content:

from ripperdoc_agent_sdk import (
    UserMessage,
    AssistantMessage,
    SystemMessage,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    ThinkingBlock
)

async for message in client.receive_messages():
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(f"Text: {block.text}")
            elif isinstance(block, ThinkingBlock):
                print(f"Thinking: {block.thinking}")
            elif isinstance(block, ToolUseBlock):
                print(f"Tool: {block.name}")

    elif isinstance(message, ResultMessage):
        print(f"Usage: {message.result.usage}")

Content Blocks

Messages contain content blocks of different types:

  • TextBlock: Plain text response
  • ThinkingBlock: Extended thinking output (for models with thinking enabled)
  • ToolUseBlock: Tool being called with parameters
  • ToolResultBlock: Result from a tool execution

MCP Servers

Integrate external MCP servers for additional tools:

from ripperdoc_agent_sdk import RipperdocAgentOptions, McpServerConfig

options = RipperdocAgentOptions(
    mcp_servers={
        "filesystem": McpServerConfig(
            type="stdio",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/allowed/path"]
        ),
        "postgres": McpServerConfig(
            type="stdio",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-postgres"],
            env={"POSTGRES_CONNECTION_STRING": "postgresql://..."}
        )
    }
)

SDK MCP Server (In-Process Tools)

Define custom tools that run in-process within your Python application:

from ripperdoc_agent_sdk import (
    RipperdocAgentOptions,
    query,
    tool,
    create_sdk_mcp_server
)

@tool("add", "Add two numbers", {"a": float, "b": float})
async def add(args):
    result = args["a"] + args["b"]
    return {"content": [{"type": "text", "text": f"Sum: {result}"}]}

@tool("multiply", "Multiply two numbers", {"a": float, "b": float})
async def multiply(args):
    result = args["a"] * args["b"]
    return {"content": [{"type": "text", "text": f"Product: {result}"}]}

# Create SDK MCP server
calculator_server = create_sdk_mcp_server(
    name="calculator",
    version="1.0.0",
    tools=[add, multiply]
)

# Use with Ripperdoc
options = RipperdocAgentOptions(
    mcp_servers={"calculator": calculator_server},
    allowed_tools=["add", "multiply"]
)

async for message in query(
    prompt="What is 25 * 17 + 10?",
    options=options
):
    if message.type == "assistant":
        for block in message.content:
            if block.type == "text":
                print(block.text)

Programmatic Agents

Define custom subagents programmatically:

from ripperdoc_agent_sdk import RipperdocAgentOptions, AgentConfig

options = RipperdocAgentOptions(
    agents={
        "code-reviewer": AgentConfig(
            description="Review code for security issues and best practices",
            prompt="You are a security expert. Focus on identifying vulnerabilities, "
                   "injection attacks, and adherence to security best practices.",
            tools=["Read", "Grep", "Glob"],
            model="sonnet"
        ),
        "test-writer": AgentConfig(
            description="Write comprehensive tests for code",
            prompt="You are a testing specialist. Write thorough unit tests "
                   "with good coverage and edge cases.",
            tools=["Read", "Write", "Glob", "Grep"],
            model="haiku"
        )
    }
)

Hooks

Intercept and modify events during execution:

from ripperdoc_agent_sdk import RipperdocAgentOptions, HookMatcher

async def log_tool_use(event, input_data):
    print(f"Tool used: {input_data.get('tool_name')}")
    return {}

async def sanitize_input(event, input_data):
    # Modify tool input before execution
    return {"updated_input": sanitize(input_data)}

options = RipperdocAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(callback=log_tool_use, tool_pattern="*"),
            HookMatcher(callback=sanitize_input, tool_pattern="Bash*")
        ]
    }
)

Hook Events

Available hook events:

  • PreToolUse - Before a tool is called
  • PostToolUse - After a tool completes
  • PostToolUseFailure - After a tool fails or times out
  • UserPromptSubmit - When user submits a prompt
  • Stop - When a stop is requested
  • SubagentStart - Before a subagent starts
  • SubagentStop - When a subagent stops
  • PreCompact - Before conversation compaction
  • Setup - Repository setup/maintenance (first startup per project)

Session Management

Control conversation lifecycle:

from ripperdoc_agent_sdk import RipperdocSDKClient, RipperdocAgentOptions

options = RipperdocAgentOptions(
    max_turns=10,  # Limit conversation to 10 turns
)

# Resume from existing session
options = RipperdocAgentOptions(
    resume="session-id-here"
)

# Continue most recent conversation
options = RipperdocAgentOptions(
    continue_conversation=True
)

# Fork a new branch when resuming
options = RipperdocAgentOptions(
    resume="session-id-here",
    fork_session=True
)

async with RipperdocSDKClient(options=options) as client:
    print(f"Session ID: {client.session_id}")
    print(f"Turn count: {client.turn_count}")

Use Cases

Code Analysis

from ripperdoc_agent_sdk import query, RipperdocAgentOptions

async def analyze_code(code: str):
    options = RipperdocAgentOptions(
        system_prompt="You are a security expert.",
        allowed_tools=["Read"],
        permission_mode="default"
    )

    async for message in query(
        prompt=f"Analyze this code for security issues:\n\n```python\n{code}\n```",
        options=options
    ):
        if message.type == "assistant":
            for block in message.content:
                if block.type == "text":
                    return block.text

Automated Refactoring

from ripperdoc_agent_sdk import RipperdocSDKClient, RipperdocAgentOptions

async def refactor_codebase():
    options = RipperdocAgentOptions(
        permission_mode="acceptEdits",
        allowed_tools=["Read", "Glob", "Grep", "Edit"],
        max_turns=20
    )

    async with RipperdocSDKClient(options=options) as client:
        await client.query("Find all deprecated API calls")
        async for msg in client.receive_messages():
            pass  # Process results

        await client.query("Update them to use the new API")
        async for msg in client.receive_messages():
            pass  # Process results

Documentation Generation

from ripperdoc_agent_sdk import query, RipperdocAgentOptions

async def generate_docs():
    options = RipperdocAgentOptions(
        allowed_tools=["Read", "Glob", "Grep"],
        agents={
            "doc-writer": AgentConfig(
                description="Generate API documentation",
                prompt="You are a technical writer. Generate clear, concise documentation.",
            )
        }
    )

    async for message in query(
        prompt="Generate API documentation for all public functions in src/api/",
        options=options
    ):
        if message.type == "assistant":
            for block in message.content:
                if block.type == "text":
                    print(block.text)

SDK Features

  • Async/await support: Native async Python API
  • Type safety: Full type hints and Pydantic models
  • Tool access: Use any Ripperdoc tool
  • Session management: Maintain conversation context
  • Permission control: Four permission modes with custom checkers
  • Streaming: Real-time message streaming
  • Thinking mode: Extended reasoning support
  • MCP integration: Connect external MCP servers
  • SDK MCP servers: Define in-process tools
  • Programmatic agents: Dynamic agent definitions
  • Hooks: Intercept and modify execution
  • Subprocess architecture: Clean separation via JSON Control Protocol

Next Steps