Python SDK

uv add raysurfer

export RAYSURFER_API_KEY=your_api_key_here

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

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
)

from raysurfer import AsyncRaySurfer

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

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

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)

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

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

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

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

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

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

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

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