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-sdkQuick 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:
| Mode | Description |
|---|---|
default | Prompts for dangerous operations (bash, file edits, etc.) |
acceptEdits | Auto-accept file edits, prompt for other dangerous operations |
bypassPermissions | Allow all operations without prompting |
plan | Planning 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 calledPostToolUse- After a tool completesPostToolUseFailure- After a tool fails or times outUserPromptSubmit- When user submits a promptStop- When a stop is requestedSubagentStart- Before a subagent startsSubagentStop- When a subagent stopsPreCompact- Before conversation compactionSetup- 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.textAutomated 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 resultsDocumentation 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
- SDK Client Reference - Detailed API reference
- TypeScript SDK - TypeScript/JavaScript SDK
- MCP Integration - Model Context Protocol
- Hooks - Event hooks and customization