Skip to Content
SDKEntitiesEndpoints

Endpoints

An endpoint is an LLM application or API that tests run against. It defines how Rhesis connects to your application, sends test inputs, and receives responses for evaluation.

For code-first endpoint registration using decorators, see the Connector documentation.

Endpoints use the shared entity interface for fetching, filtering, updating, and deleting.

Properties

PropertyTypeDescription
idstrUnique identifier
namestrDisplay name
descriptionstrEndpoint description
connection_typeConnectionTypeHow Rhesis connects: REST, WebSocket, GRPC, or SDK
urlstrEndpoint URL (for REST/WebSocket/GRPC)
project_idstrAssociated project ID

name, connection_type, and project_id are required to save an endpoint. REST endpoints also carry request/response mapping, headers, and an auth_token for the target API — see the Connector for those.

Connection types

TypeDescription
RESTHTTP API endpoints
WebSocketReal-time WebSocket connections
GRPCgRPC service endpoints
SDKFunctions registered via the Connector decorator

Fetching endpoints

fetch_endpoints.py
from rhesis.sdk.entities import Endpoints

for endpoint in Endpoints.all():
    print(f"{endpoint.name} ({endpoint.connection_type})")

# By name or ID
endpoint = Endpoints.pull(name="Production Chatbot")

Invoking endpoints

Send an input and receive a response. Pass conversation_id to continue a multi-turn conversation (session_id is a deprecated alias):

invoke_endpoint.py
endpoint = Endpoints.pull(name="Customer Support Bot")

# Single turn
response = endpoint.invoke(input="What are your business hours?")
print(response["output"])

# Multi-turn: reuse the conversation_id returned by the previous call
first = endpoint.invoke(input="I need help with my order")
second = endpoint.invoke(
    input="It was order #12345",
    conversation_id=first["conversation_id"],
)

invoke() returns a standardized response:

response_structure.py
{
    "output": "Response text from the endpoint",
    "conversation_id": "conversation-abc",
    "context": [...],    # Optional retrieval context (e.g. RAG sources)
    "metadata": {...},   # Optional structured metadata
}

Testing connectivity

test() sends a short probe and raises ValueError if the endpoint does not respond:

test_endpoint.py
endpoint = Endpoints.pull(name="Production Chatbot")

try:
    endpoint.test()
except ValueError as e:
    print(f"Endpoint not responding: {e}")

Creating endpoints

create_endpoint.py
from rhesis.sdk.entities import Endpoint
from rhesis.sdk.entities.endpoint import ConnectionType

endpoint = Endpoint(
    name="Staging API",
    connection_type=ConnectionType.REST,
    url="https://staging-api.example.com/chat",
    project_id="project-123",
)
endpoint.push()

Running tests against an endpoint

run_tests.py
from rhesis.sdk.entities import TestSets, Tests, Endpoints

endpoint = Endpoints.pull(name="Production Chatbot")

# A whole test set
result = TestSets.pull(name="Safety Evaluation").execute(endpoint)

# A single test
result = Tests.pull(id="test-123").execute(endpoint)

SDK-registered endpoints

Functions decorated with @endpoint via the Connector appear as SDK endpoints and are fetched like any other:

sdk_endpoint.py
from rhesis.sdk.entities import Endpoints

sdk_endpoints = Endpoints.all(filter="connection_type eq 'SDK'")
for ep in sdk_endpoints:
    print(ep.name)

Next: register a code-first endpoint with the Connector, then run a test set against it.