Python SDK

Installation

uv add raysurfer

Setup

Set your API key:

export RAYSURFER_API_KEY=your_api_key_here

Get your key from the dashboard.

Usage

Swap your client class and method names. Options come directly from claude_agent_sdk:

# Before
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

# After
from raysurfer import RaysurferClient
from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Bash"],
    system_prompt="You are a helpful assistant.",
)

async with RaysurferClient(options) as client:
    await client.query("Generate quarterly report")
    async for msg in client.response():
        print(msg)

Method Mapping

Claude SDK	Raysurfer
`ClaudeSDKClient(options)`	`RaysurferClient(options)`
`await client.query(prompt)`	`await client.query(prompt)`
`client.receive_response()`	`client.response()`

Options

Options are passed through directly from claude_agent_sdk.ClaudeAgentOptions. All standard options work:

from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Bash"],
    system_prompt="You are a helpful assistant.",
    model="claude-opus-4-6",
    max_turns=10,
    cwd="/path/to/working/dir",
    # ... all other ClaudeAgentOptions fields
)

Caching is enabled automatically when RAYSURFER_API_KEY is set.

Agent-Owned Repos

Set agent_id to isolate uploads/searches into an agent-owned repo:

from raysurfer import AsyncRaySurfer

client = AsyncRaySurfer(
    api_key="your_api_key",
    agent_id="claims-agent-v1",
)

Agent-Accessible Functions

Decorator flow:

from raysurfer import AsyncRaySurfer, agent_accessible

rs = AsyncRaySurfer(api_key="your_api_key", agent_id="claims-agent-v1")

@agent_accessible("Generate claim summary PDF")
def generate_claim_summary(claim_id: str) -> str:
    return f"summary-{claim_id}.pdf"

await rs.publish_function_registry([generate_claim_summary])

raysurfer.yaml flow:

from my_app import claims
from raysurfer import AsyncRaySurfer, load_config

rs = AsyncRaySurfer(api_key="your_api_key", agent_id="claims-agent-v1")
functions = load_config("raysurfer.yaml", modules=[claims])
await rs.publish_function_registry(functions)

Snippet Retrieval Scope

Control which cached snippets are retrieved using snips_desired:

from raysurfer import RaysurferClient
from claude_agent_sdk import ClaudeAgentOptions

options = ClaudeAgentOptions(
    allowed_tools=["Read", "Write", "Bash"],
)

# Include company-level snippets
client = RaysurferClient(
    options,
    snips_desired="company",  # Company-level snippets (Team/Enterprise)
)

# Pro/Enterprise: Retrieve client-specific snippets
client = RaysurferClient(
    options,
    snips_desired="client",   # Client workspace snippets (Pro or Enterprise)
)

Configuration	Required Tier
`snips_desired="company"`	PRO or ENTERPRISE
`snips_desired="client"`	PRO or ENTERPRISE (requires `workspace_id`)

Public Snippets

Include community public snippets (crawled from GitHub) alongside your private snippets:

# High-level
client = RaysurferClient(options, public_snips=True)

# Low-level
client = RaySurfer(api_key="...", public_snips=True)

See How It Works — Public Snippets for details.

Full Example

import asyncio
import os
from raysurfer import RaysurferClient
from claude_agent_sdk import ClaudeAgentOptions

os.environ["RAYSURFER_API_KEY"] = "your_api_key"

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Write", "Bash"],
        system_prompt="You are a helpful assistant.",
    )

    async with RaysurferClient(options) as client:
        # First run: generates and caches code
        await client.query("Fetch GitHub trending repos")
        async for msg in client.response():
            print(msg)

        # Second run: retrieves from cache (instant)
        await client.query("Fetch GitHub trending repos")
        async for msg in client.response():
            print(msg)

asyncio.run(main())

Without Caching

If RAYSURFER_API_KEY is not set, RaysurferClient behaves exactly like ClaudeSDKClient — no caching, just a pass-through wrapper.

Low-Level API

For custom integrations, use the RaySurfer client directly.

Complete Low-Level Example with Anthropic API

import anthropic
from raysurfer import RaySurfer
from raysurfer.types import FileWritten, LogFile

client = RaySurfer(api_key="your_raysurfer_api_key", agent_id="claims-agent-v1")
task = "Fetch GitHub trending repos"

# 1. Search for cached code matching a task
result = client.search(
    task=task,
    top_k=5,
    min_verdict_score=0.3,
    per_function_reputation=True,
)

for match in result.matches:
    print(f"{match.code_block.name}: {match.combined_score}")
    print(f"  Source: {match.code_block.source[:80]}...")

# Your LLM execution loop goes here
# e.g. Claude Agent SDK generates and runs code using the cached snippets

# 2. Upload a new code file after execution
#    AI voting is triggered automatically when use_raysurfer_ai_voting is True (default)
file = FileWritten(path="fetch_repos.py", content="def fetch(): ...")
client.upload(
    task=task,
    file_written=file,
    succeeded=True,
    execution_logs="Fetched 10 trending repos successfully",
    dependencies={"httpx": "0.27.0", "pydantic": "2.5.0"},  # Track package versions
    per_function_reputation=True,
)

# 2b. Bulk upload prompts/logs/code for sandboxed grading
logs = [LogFile(path="logs/run.log", content="Task completed", encoding="utf-8")]
client.upload_bulk_code_snips(
    prompts=["Build a CLI tool", "Add CSV support"],
    files_written=[FileWritten(path="cli.py", content="def main(): ...")],
    log_files=logs,
)

# 3. (Optional) Vote on a cached snippet manually
#    Only needed if you want explicit control — upload already votes by default
client.vote_code_snip(
    task=task,
    code_block_id=result.matches[0].code_block.id,
    code_block_name=result.matches[0].code_block.name,
    code_block_description=result.matches[0].code_block.description,
    succeeded=True,
)

Response Fields

The search() response includes:

Field	Type	Description
`files`	`list[CodeFile]`	Retrieved code files with metadata
`task`	`str`	The task that was searched
`total_found`	`int`	Total matches found
`add_to_llm_prompt`	`str`	Pre-formatted string to append to your LLM system prompt

Each CodeFile contains:

Field	Type	Description
`code_block_id`	`str`	Unique identifier of the code block
`filename`	`str`	Generated filename (e.g., `"github_fetcher.py"`)
`source`	`str`	Full source code
`entrypoint`	`str`	Main function to call
`description`	`str`	What the code does
`input_schema`	`JsonDict`	JSON schema for inputs
`output_schema`	`JsonDict`	JSON schema for outputs
`language`	`str`	Programming language
`dependencies`	`dict[str, str]`	Package name to version mapping
`verdict_score`	`float`	Quality rating (thumbs_up / total, 0.3 if unrated)
`thumbs_up`	`int`	Positive votes
`thumbs_down`	`int`	Negative votes
`similarity_score`	`float`	Semantic similarity (0–1)
`combined_score`	`float`	Combined score (similarity * 0.6 + verdict * 0.4)

Async Version

import anthropic
from raysurfer import AsyncRaySurfer
from raysurfer.types import FileWritten

async with AsyncRaySurfer(api_key="your_api_key") as client:
    # 1. Search for cached code
    result = await client.search(
        task="Fetch GitHub trending repos",
        per_function_reputation=True,
    )

    for match in result.matches:
        print(f"{match.code_block.name}: {match.combined_score}")

    # Your LLM execution loop goes here
    # e.g. Claude Agent SDK generates and runs code using the cached snippets

    # 2. Upload a new code file after execution (voting triggered by default)
    file = FileWritten(path="fetch_repos.py", content="def fetch(): ...")
    await client.upload(
        task=task,
        file_written=file,
        succeeded=True,
        execution_logs="Fetched 10 trending repos successfully",
        per_function_reputation=True,
    )

    # 3. (Optional) Vote on snippet manually — only if you need explicit control
    await client.vote_code_snip(
        task=task,
        code_block_id=result.matches[0].code_block.id,
        code_block_name=result.matches[0].code_block.name,
        code_block_description=result.matches[0].code_block.description,
        succeeded=True,
    )

Search Response

The search() method returns a SearchResponse:

Field	Type	Description
`matches`	`list[SearchMatch]`	Matching code blocks with full scoring
`total_found`	`int`	Total matches found
`cache_hit`	`bool`	Whether results were served from cache
`search_namespaces`	`list[str]`	Namespaces that were searched

Each SearchMatch contains:

Field	Type	Description
`code_block`	`CodeBlock`	The code block (has `id`, `name`, `description`, `source`, `entrypoint`, `language`, `dependencies`, `tags`)
`combined_score`	`float`	Overall relevance score (0–1)
`vector_score`	`float`	Semantic similarity score
`verdict_score`	`float`	Community quality rating
`error_resilience`	`float`	Error handling robustness score
`thumbs_up`	`int`	Positive votes
`thumbs_down`	`int`	Negative votes
`filename`	`str`	Generated filename for the code
`language`	`str`	Programming language
`entrypoint`	`str`	Main function name
`dependencies`	`dict[str, str]`	Package name to version mapping (e.g., `{"pandas": "2.1.0"}`)
`agent_id`	`str \| None`	Agent identifier when snippet came from an agent-owned repo
`s3_url`	`str \| None`	Presigned download URL for agent-stored script content
`functions`	`list[FunctionReputation] \| None`	Per-function reputation metadata (when `per_function_reputation=True`)

Method Reference

Method	Description
`search(task, top_k, min_verdict_score, prefer_complete, per_function_reputation)`	Search for cached code snippets
`upload(task, file_written, succeeded, ..., per_function_reputation=False)`	Store a code file for future reuse
`upload_bulk_code_snips(prompts, files_written, ...)`	Bulk upload for grading
`delete(snippet_id)`	Delete a snippet by ID or name
`vote_code_snip(task, code_block_id, name, description, succeeded)`	Vote on snippet usefulness
`comment_on_code_snip(code_block_id, text)`	Add a comment to a snippet
`publish_function_registry(functions)`	Publish `@agent_accessible` functions for agent discovery
`tool(fn)`	Register a local function as a tool for `execute()`
`execute(task, user_code=None, timeout=300, codegen_api_key=None, codegen_prompt=None, codegen_model=...)`	Execute user code (primary) or sandbox-generated code (optional) with registered tools

Programmatic Tool Calling

Also available as an integration guide: Register Python Functions. Register local tools, then either pass user_code (primary mode) or use optional sandbox codegen with your own key + prompt:

import asyncio
from raysurfer import AsyncRaySurfer

async def main():
    rs = AsyncRaySurfer(api_key="your_api_key")

    @rs.tool
    def add(a: int, b: int) -> int:
        """Add two numbers together."""
        return a + b

    @rs.tool
    def multiply(a: int, b: int) -> int:
        """Multiply two numbers together."""
        return a * b

    user_code = """
intermediate = add(5, 3)
final = multiply(intermediate, 2)
print(final)
"""
    result = await rs.execute(
        "Add 5 and 3, then multiply the result by 2",
        user_code=user_code,
    )
    print(result.result)       # "16"
    print(result.tool_calls)   # [ToolCallRecord(...), ...]
    print(result.cache_hit)    # False (reserved field)

asyncio.run(main())

The @rs.tool decorator can wrap any Python function you want to expose as a tool. It introspects the function signature to build a JSON schema. Your function docstring becomes the tool description field in the tool schema payload. Both sync and async callbacks are supported.

How It Works

SDK opens a live callback channel for tool call routing
Your app sends either user_code (primary mode) or codegen_* inputs (optional mode) to /api/execute/run
Code runs in a Modal sandbox — tool calls are routed back to your local functions through that callback channel
Results are returned with full tool call history

Use exactly one mode per call:

user_code: run pre-generated code directly (recommended default)
codegen_api_key + codegen_prompt: generate code in sandbox and run it

Execute Options

result = await rs.execute(
    "Your task description",
    user_code="print(add(1, 2))", # Primary mode
    timeout=300,            # Max execution time in seconds (default 300)
)

result = await rs.execute(
    "Your task description",
    codegen_api_key="YOUR_ANTHROPIC_API_KEY",  # Optional mode
    codegen_prompt="Write Python code that uses add(a, b) and prints 2 + 3.",
    codegen_model="claude-opus-4-6",
    timeout=300,
)

ExecuteResult Fields

Field	Type	Description
`execution_id`	`str`	Unique execution identifier
`result`	`str \| None`	Stdout output from the script
`exit_code`	`int`	Process exit code (0 = success)
`duration_ms`	`int`	Total execution time
`cache_hit`	`bool`	Reserved field (currently always `False` for execute)
`error`	`str \| None`	Error message if exit_code != 0
`tool_calls`	`list[ToolCallRecord]`	All tool calls made during execution

Get Started

Integrations

SDK Reference

Enterprise

Rate Limits & Pricing

Installation

Setup

Usage

Method Mapping

Options

Agent-Owned Repos

Agent-Accessible Functions

Snippet Retrieval Scope

Public Snippets

Full Example

Without Caching

Low-Level API

Complete Low-Level Example with Anthropic API

Response Fields

Async Version

Search Response

Method Reference

Programmatic Tool Calling

How It Works

Execute Options

ExecuteResult Fields

​Installation

​Setup

​Usage

​Method Mapping

​Options

​Agent-Owned Repos

​Agent-Accessible Functions

​Snippet Retrieval Scope

​Public Snippets

​Full Example

​Without Caching

​Low-Level API

​Complete Low-Level Example with Anthropic API

​Response Fields

​Async Version

​Search Response

​Method Reference

​Programmatic Tool Calling

​How It Works

​Execute Options

​ExecuteResult Fields

Installation

Setup

Usage

Method Mapping

Options

Agent-Owned Repos

Agent-Accessible Functions

Snippet Retrieval Scope

Public Snippets

Full Example

Without Caching

Low-Level API

Complete Low-Level Example with Anthropic API

Response Fields

Async Version

Search Response

Method Reference

Programmatic Tool Calling

How It Works

Execute Options

ExecuteResult Fields