SDK Endpoints

Register Python functions as testable endpoints using the Rhesis SDK. Instead of configuring endpoints manually in the dashboard, you decorate your functions and the SDK handles registration, mapping, and synchronization automatically.

How It Works

The SDK connector establishes a WebSocket connection to the Rhesis backend. When your application starts, the SDK:

1. Discovers all functions decorated with @endpoint
2. Registers them as endpoints in your Rhesis project
3. Keeps the connection alive for remote invocation during tests

When Rhesis runs tests against an SDK endpoint, it sends the test input over the WebSocket connection, the SDK calls your function locally, and returns the result to the platform for evaluation.

Quick Start

from rhesis.sdk import RhesisClient

client = RhesisClient(
    api_key="your-api-key",
    project_id="your-project-id",
    environment="development"
)

from rhesis.sdk import endpoint

@endpoint()
def chat(input: str, conversation_id: str = None) -> dict:
    """Handle chat messages."""
    response = generate_response(input)
    return {
        "output": response,
        "conversation_id": conversation_id or generate_conversation_id(),
    }

Auto-Mapping

When your function uses standard field names, the SDK automatically maps them to platform variables. No manual mapping configuration is needed.

Standard request fields: input, conversation_id (or legacy session_id), context, metadata, tool_calls

Standard response fields: output, context, metadata, tool_calls, conversation_id (or legacy session_id)

@endpoint()
def chat(input: str, conversation_id: str = None) -> dict:
    """Standard names are auto-detected."""
    return {
        "output": process_message(input),
        "conversation_id": conversation_id,
    }

Manual Mapping

For functions with custom parameter names or complex structures, provide explicit mappings using the same Jinja2 and JSONPath syntax as REST endpoint mappings:

@endpoint(
    request_mapping={
        "user_query": "{{ input }}",
        "conv_id": "{{ conversation_id }}",
    },
    response_mapping={
        "output": "$.result.text",
        "conversation_id": "$.conv_id",
    },
)
def chat(user_query: str, conv_id: str = None) -> dict:
    """Custom names require manual mapping."""
    return {
        "result": {"text": process(user_query)},
        "conv_id": conv_id,
    }

Multiple Functions

Register multiple endpoints from a single application. Each decorated function becomes its own endpoint:

@endpoint()
def handle_chat(input: str, conversation_id: str = None) -> dict:
    """Process chat messages."""
    return {"output": generate_response(input), "conversation_id": conversation_id}

@endpoint()
def search_documents(input: str) -> dict:
    """Search documents."""
    results = perform_search(input)
    return {"output": format_results(results), "context": results}

@endpoint()
def summarize(input: str) -> dict:
    """Summarize text."""
    return {"output": generate_summary(input)}

Parameter Binding

Inject infrastructure dependencies (database connections, configuration, auth context) without exposing them in the remote function signature:

@endpoint(
    bind={
        "db": lambda: get_db_session(),
        "config": AppConfig(),
    }
)
def query_data(db, config, input: str) -> dict:
    """
    db and config are injected automatically.
    Remote signature: query_data(input: str)
    """
    results = db.query(config.table, input)
    return {"output": format_results(results)}

For full details on binding patterns, see Parameter Binding.

Parameter-aware Endpoints

When your project uses Parameters & Experiments, map resolved parameter values from params into your function arguments with request_mapping. This is the same {{ params.* }} path used by REST endpoints, so SDK and REST endpoints behave consistently during test runs.

from rhesis.sdk import endpoint

@endpoint(
    name="chat",
    request_mapping={
        "input": "{{ input }}",
        "model": "{{ params.model | default('gpt-4o') }}",
        "temperature": "{{ params.temperature | default(0.7) }}",
        "system_prompt": "{{ params.system_prompt | default('You are helpful.') }}",
    },
    response_mapping={"output": "$.answer"},
)
def chat(input: str, *, model: str, temperature: float, system_prompt: str) -> dict:
    answer = run_chat(input, model=model, temperature=temperature, system_prompt=system_prompt)
    return {"answer": answer}

For more details on request-mapping and context-resolved injection, see Connector Injection.

SDK vs. REST Endpoints

Connection Management

• Auto-reconnect: The SDK reconnects automatically with exponential backoff if the connection is lost.
• Re-registration: Functions are re-registered on reconnect.
• Status sync: Adding or modifying functions updates endpoints automatically. Removing a function marks its endpoint as Inactive.

Disabling the Connector

To disable all connector functionality (useful for CI/CD or testing), set the environment variable:

export RHESIS_CONNECTOR_DISABLED=true

When disabled, @endpoint decorators return functions unmodified and no WebSocket connection is established.

Learn more

• SDK Connector — full reference with advanced patterns
• Input/Output Mapping — mapping guide
• Parameter Binding — dependency injection patterns
• Advanced Mapping — complex type serialization
• Examples — complete working examples