raysurfer 0.6.12

Semantic code caching for LLM agents

RaySurfer Python SDK

LLM output caching for AI agents. Retrieve proven code instead of regenerating it.

Installation

pip install raysurfer

Setup

Set your API key:

export RAYSURFER_API_KEY=your_api_key_here

Get your key from the dashboard.

Low-Level API

For custom integrations, use the RaySurfer client directly with any LLM provider.

Complete Example with Anthropic API

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

client = RaySurfer(api_key="your_raysurfer_api_key")
task = "Fetch GitHub trending repos"

# 1. Retrieve cached code files for a task
result = client.get_code_files(
    task=task,
    top_k=5,
    min_verdict_score=0.3,
)

# result.add_to_llm_prompt contains a pre-formatted string like:
#
# You have access to pre-written code files:
# - .raysurfer_code/github_fetcher.py: Fetches trending repos
# ...

# Augment your system prompt with cached code context
base_prompt = "You are a helpful coding assistant."
augmented_prompt = base_prompt + result.add_to_llm_prompt

# Make your Anthropic API call with the augmented prompt
anthropic_client = anthropic.Anthropic()
response = anthropic_client.messages.create(
    model="claude-opus-4-5-20250514",
    max_tokens=1024,
    system=augmented_prompt,
    messages=[{"role": "user", "content": "Fetch the top 10 trending GitHub repos"}],
)

print(response.content[0].text)

# 2. Upload a new code file after execution
file = FileWritten(path="fetch_repos.py", content="def fetch(): ...")
client.upload_new_code_snip(
    task=task,
    file_written=file,
    succeeded=True,
    execution_logs="Fetched 10 trending repos successfully",
    dependencies={"httpx": "0.27.0", "pydantic": "2.5.0"},
)

# 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. Vote on whether a cached snippet was useful
client.vote_code_snip(
    task=task,
    code_block_id=result.files[0].code_block_id,
    code_block_name=result.files[0].filename,
    code_block_description=result.files[0].description,
    succeeded=True,
)

Async Version

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

async with AsyncRaySurfer(api_key="your_api_key") as client:
    # 1. Retrieve cached code
    result = await client.get_code_files(task="Fetch GitHub trending repos")

    # 2. Use add_to_llm_prompt to augment your LLM call
    augmented_prompt = "You are helpful." + result.add_to_llm_prompt

    # 3. Upload a new code file after execution
    file = FileWritten(path="fetch_repos.py", content="def fetch(): ...")
    await client.upload_new_code_snip(
        task="Fetch GitHub trending repos",
        file_written=file,
        succeeded=True,
        execution_logs="Fetched 10 trending repos successfully",
    )

    # 4. Vote on snippet manually
    await client.vote_code_snip(
        task="Fetch GitHub trending repos",
        code_block_id=result.files[0].code_block_id,
        code_block_name=result.files[0].filename,
        code_block_description=result.files[0].description,
        succeeded=True,
    )

Client Options

client = RaySurfer(
    api_key="your_api_key",
    base_url="https://api.raysurfer.com",  # optional
    timeout=30,                             # optional, in seconds
    organization_id="org_xxx",              # optional, for team namespacing
    workspace_id="ws_xxx",                  # optional, for enterprise namespacing
    snips_desired="company",                # optional, snippet scope
    public_snips=True,                      # optional, include community snippets
)

Response Fields

The get_code_files() 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 code_block_id, filename, source, description, verdict_score, thumbs_up, thumbs_down, and similarity_score.

Store a Code Block with Full Metadata

result = client.store_code_block(
    name="GitHub User Fetcher",
    source="def fetch_user(username): ...",
    entrypoint="fetch_user",
    language="python",
    description="Fetches user data from GitHub API",
    tags=["github", "api", "user"],
    dependencies={"httpx": "0.27.0", "pydantic": "2.5.0"},
)

Retrieve Few-Shot Examples

examples = client.get_few_shot_examples(task="Parse CSV files", k=3)

for ex in examples:
    print(f"Task: {ex.task}")
    print(f"Code: {ex.code_snippet}")

Retrieve Task Patterns

patterns = client.get_task_patterns(
    task="API integration",
    min_thumbs_up=5,
    top_k=20,
)

for p in patterns:
    print(f"{p.task_pattern} -> {p.code_block_name}")

User-Provided Votes

Instead of relying on AI voting, provide your own votes:

# Single upload with your own vote (AI voting is skipped)
client.upload_new_code_snip(
    task="Fetch GitHub trending repos",
    file_written=file,
    succeeded=True,
    user_vote=1,  # 1 = thumbs up, -1 = thumbs down
)

# Bulk upload with per-file votes (AI grading is skipped)
client.upload_bulk_code_snips(
    prompts=["Build a CLI tool", "Add CSV support"],
    files_written=files,
    log_files=logs,
    user_votes={
        "app.py": 1,     # thumbs up
        "utils.py": -1,  # thumbs down
    },
)

Method Reference

Method	Description
search(task, top_k, min_verdict_score, prefer_complete, input_schema)	Unified search for cached code (recommended)
get_code_files(task, top_k, min_verdict_score, prefer_complete, cache_dir)	Retrieve cached code files with add_to_llm_prompt for LLM augmentation
get_code_snips(task, top_k, min_verdict_score)	Retrieve cached code snippets by semantic search
retrieve_best(task, top_k, min_verdict_score)	Retrieve the single best match
get_few_shot_examples(task, k)	Retrieve few-shot examples for code generation prompting
get_task_patterns(task, min_thumbs_up, top_k)	Retrieve proven task-to-code mappings
store_code_block(name, source, entrypoint, language, description, tags, dependencies, ...)	Store a code block with full metadata
upload_new_code_snip(task, file_written, succeeded, use_raysurfer_ai_voting, user_vote, execution_logs, dependencies)	Store a single code file with optional dependency versions
upload_bulk_code_snips(prompts, files_written, log_files, use_raysurfer_ai_voting, user_votes)	Bulk upload for grading (AI votes by default, or provide per-file votes)
vote_code_snip(task, code_block_id, name, description, succeeded)	Vote on snippet usefulness

Exceptions

Both sync and async clients include built-in retry logic with exponential backoff for transient failures (429, 5xx, network errors).

Exception	Description
RaySurferError	Base exception for all Raysurfer errors
APIError	API returned an error response (includes status_code)
AuthenticationError	API key is invalid or missing
CacheUnavailableError	Cache backend is unreachable
RateLimitError	Rate limit exceeded after retries (includes retry_after)
ValidationError	Request validation failed (includes field)

from raysurfer import RaySurfer
from raysurfer.exceptions import RateLimitError

client = RaySurfer(api_key="your_api_key")

try:
    result = client.get_code_snips(task="Fetch GitHub repos")
except RateLimitError as e:
    print(f"Rate limited after retries: {e}")
    if e.retry_after:
        print(f"Try again in {e.retry_after}s")

---

Claude Agent SDK Drop-in

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()

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.

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)
)

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

Configuration	Required Tier
snips_desired="company"	TEAM or ENTERPRISE
snips_desired="client"	ENTERPRISE only

Public Snippets

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

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

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

License

MIT