---
url: https://docs.rhesis.ai/sdk/connector
title: Connector
---
# 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](/docs/endpoints).

**`@endpoint` vs `@observe`:** Tracing-only helpers use `@observe`; functions Rhesis should **invoke from the platform** (test runs, connector) use `@endpoint`. See [Observe vs endpoint](/docs/tracing/decorators#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

### Initialize the Client

```python
from rhesis.sdk import RhesisClient

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

### Decorate Functions

```python
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(),
    }
```

### Automatic Registration

When your app starts, functions are automatically registered as endpoints. View them in **Projects** → **Your Project** → **Endpoints**.

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

```python
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, do not call `connect()` — it raises. Call
`client.start_connector()` instead, from wherever that loop starts.

For a web app this is not optional. `@endpoint` registers your function when the module is
*imported*, and uvicorn imports it before it creates the event loop, so the connector has no loop
to dial from and defers the connection. Nothing retries afterwards: the SDK logs a warning and your
endpoint never reaches the platform. Start it from a FastAPI `lifespan`:

```python
from contextlib import asynccontextmanager

from fastapi import FastAPI
from rhesis.sdk import RhesisClient, endpoint

client = RhesisClient.from_environment()

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

@asynccontextmanager
async def lifespan(app: FastAPI):
    # The loop exists by now, so the connector can dial out.
    client.start_connector()
    yield

app = FastAPI(lifespan=lifespan)
```

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

```python
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:

| Attribute or method | Description |
| --- | --- |
| `organization_id` | Organization associated with the connector execution |
| `user_id` | User associated with the connector execution |
| `project_id` | Active project for this execution, if available |
| `get_db()` | Returns a context-managed database session scoped to the tenant |

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.

```python
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.

| Parameter | Required | Description |
|---|---|---|
| `input` | Yes | Input prompt or user message |
| `output` | Yes | Model output to evaluate |
| `expected_output` | No | Ground truth reference output |
| `context` | No | Retrieved context documents used for generation |

> The metric function must return a dict containing at least a `score` key (or a `MetricResult` instance).
>   Any unsupported parameter names cause registration-time errors.

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

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

> `@endpoint` registrations still require a `project_id`. It can come from the constructor, `RHESIS_PROJECT_ID`
>   env var, or a project-scoped API token. The optional `project_id` behavior applies to metrics-only connector usage.

### See it in action

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

[Watch on YouTube](https://www.youtube.com/watch?v=6hteHCpOxpM)

## Environment

The `environment` parameter is **required** and must be one of:

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

> **Production**: Changes take effect immediately. Test in development/staging first.

```python
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:

```bash
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 after the function, and status `Active` (connected) or `Inactive` (disconnected). `Active` means the function is registered and reachable, not that its mapping works: registration never calls the function, so a mapping error shows up on the first test run or Playground message. 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](/docs/endpoints) for external APIs or services outside your codebase. Always add type hints so parameters and return values serialize correctly (see [Advanced Mapping](./serializers)).

## 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.
- `"Connector registered with no running event loop, so it is not connected"` means a web app never started the connector. Call `client.start_connector()` from a startup hook; see [Running the connector](#running-the-connector).
- `"TLS certificate verification failed"` does not retry. Check `SSL_CERT_FILE` / `SSL_CERT_DIR` if you set either, add the root CA of any corporate proxy that re-signs TLS traffic, and check the system clock. An interpreter with no CA bundle of its own (for example a python.org macOS install) falls back to `certifi` for the connector automatically.

## Next steps

- [Map inputs and outputs](./mapping) to your function's parameters and return values.
- [Inject dependencies](./binding) like database sessions with `bind`.
- [Attach files](./files) to test executions.