Connector

Register Python functions as testable endpoints in Rhesis using the SDK connector. This code-first approach automatically creates and manages endpoints, providing an alternative to manual endpoint configuration.

@endpoint vs @observe: Tracing-only helpers use @observe; functions Rhesis should invoke from the platform (test runs, connector) use @endpoint. See Observe vs endpoint.

The SDK connects over a WebSocket and keeps registered endpoints in sync with your code as functions are added, changed, or removed.

Quick Start

from rhesis.sdk import RhesisClient

client = RhesisClient(
    api_key="your-api-key",
    project_id="your-project-id",
    environment="development"  # Required: "development", "staging", or "production"
)

from rhesis.sdk import endpoint

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

Running the connector

For a script that only defines endpoints and does not run a web server, call client.connect() at the end of the script. It blocks until the process is interrupted (e.g. Ctrl+C). While the script is running, you can run tests from the Rhesis platform against your registered endpoint(s).

from rhesis.sdk import RhesisClient, endpoint

client = RhesisClient.from_environment()

@endpoint()
def chat(input: str, conversation_id: str = None) -> dict:
    """Handle chat messages."""
    return {
        "output": "Echo: " + input,
        "conversation_id": conversation_id or "default-conversation",
    }

if __name__ == "__main__":
    client.connect()

If you already have a running event loop (e.g. in an async REPL or Jupyter), do not call connect(). Run the connector as a task in that loop instead.

Platform context with EndpointContext

Most SDK endpoint functions only need business inputs such as input, files, or conversation_id. Backend-integrated endpoint functions can also declare an EndpointContext parameter when they need tenant-scoped platform access during execution.

from sqlalchemy import text

from rhesis.sdk import EndpointContext, endpoint

@endpoint()
def enrich_with_project_data(input: str, ctx: EndpointContext) -> dict:
    with ctx.get_db() as db:
        # The session carries the organization, user, and project scope
        # injected by the platform connector execution context.
        scoped_count = db.execute(text("select 1")).scalar_one()

    return {
        "output": f"Scoped DB check {scoped_count}: {input}",
    }

EndpointContext is injected by type annotation. It is never accepted from remote inputs, so callers cannot forge organization_id, user_id, or project_id through the request payload. The context exposes:

When the endpoint runs inside the Rhesis backend, get_db() uses the backend tenant-aware session factory. If you construct EndpointContext outside the backend process, pass _db_factory so get_db() can create a scoped session.

SDK-side metrics with @metric

Register custom metric functions through the same connector runtime used by @endpoint. Metric functions are executed from the backend during evaluation.

from rhesis.sdk import RhesisClient, endpoint, metric

client = RhesisClient.from_environment()

@endpoint()
def chat(input: str, conversation_id: str | None = None) -> dict:
    return {
        "output": f"Echo: {input}",
        "conversation_id": conversation_id or "default-conversation",
    }

@metric(name="groundedness_check", score_type="binary")
def groundedness_check(input: str, output: str, context: list[str] | None = None) -> dict:
    grounded = bool(context) and any(source in output for source in context)
    return {
        "score": 1.0 if grounded else 0.0,
        "details": {"reason": "Checks whether output uses provided context"},
    }

if __name__ == "__main__":
    client.connect()

Allowed metric function parameters

@metric validates function signatures at registration time.

Metrics-only connector mode (optional project_id)

If your script only registers SDK metrics with @metric, project_id is optional. In this mode, the connector still establishes a WebSocket session and registers metrics for backend-side evaluation.

from rhesis.sdk import RhesisClient, metric

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

@metric(name="contains_policy_reference", score_type="binary")
def contains_policy_reference(input: str, output: str, context: list[str] | None = None) -> dict:
    has_reference = bool(context) and any(ref in output for ref in context)
    return {"score": 1.0 if has_reference else 0.0}

if __name__ == "__main__":
    client.connect()

See it in action

This video connects an LLM application with a single @endpoint decorator and runs single- and multi-turn tests against it:

Environment

The environment parameter is required and must be one of:

• development: Local iteration and testing
• staging: Pre-production validation
• production: Live systems

import os

client = RhesisClient(
    api_key=os.getenv("RHESIS_API_KEY"),
    project_id=os.getenv("RHESIS_PROJECT_ID"),
    environment=os.getenv("RHESIS_ENVIRONMENT", "development"),
)

Disabling the Connector

To disable all connector and tracing functionality (useful for CI/CD or testing), set:

export RHESIS_CONNECTOR_DISABLED=true

Accepted values: true, 1, yes, on (case-insensitive)

When disabled, @endpoint and @observe return functions unmodified, no WebSocket connection is established, and all SDK method calls become no-ops.

Viewing endpoints

Registered endpoints appear under Projects → Your Project → Endpoints with connection type SDK, named {Project Name} ({function_name}), and status Active (connected) or Inactive (disconnected). The SDK reconnects and re-registers functions automatically after a dropped connection; removing a function marks its endpoint Inactive.

When to use the connector

Use the connector when the functions under test live in your codebase and you want a code-first definition. Use manual endpoint configuration for external APIs or services outside your codebase. Always add type hints so parameters and return values serialize correctly (see Advanced Mapping).

Troubleshooting

Functions not appearing as Active:

• Confirm RhesisClient is initialized with api_key and environment, and that a project_id is available (constructor, RHESIS_PROJECT_ID env var, or a project-scoped token). Metrics-only scripts do not need project_id.
• Confirm functions use the @endpoint() decorator, then restart the app to re-register.
• "RhesisClient not initialized" means the RhesisClient instance must be created before the decorator runs.

Next steps

• Map inputs and outputs to your function’s parameters and return values.
• Inject dependencies like database sessions with bind.
• Attach files to test executions.