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.

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, session_id: str = None) -> dict:
    """Handle chat messages."""
    return {
        "output": process_message(input),
        "session_id": session_id or generate_session_id(),
    }

See It in Action

Watch this video to see how the Rhesis SDK connector integrates an LLM application in under a minute using a single decorator:

What You’ll Learn:

• Integrating an LLM app with the @endpoint decorator
• Automatic session management in action
• Single-turn and multi-turn testing made easy

No complex APIs. No orchestration. Just one decorator connecting your Python functions to comprehensive testing.

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

Mapping

Auto-Mapping (Recommended)

Use standard field names for automatic detection:

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

Standard fields:

• Request: input, session_id, context, metadata, tool_calls
• Response: output, context, metadata, tool_calls, session_id

Manual Mapping

For custom parameter names or complex structures, provide explicit mappings:

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

Request Mapping (Jinja2 Templates):

• Use Jinja2 template syntax: {{ variable_name }}
• Maps standard Rhesis request fields (input, session_id, context) to your function parameters
• Custom fields from the request are passed through automatically
• Example: "user_message": "{{ input }}" maps the Rhesis input field to your function’s user_message parameter

Response Mapping (JSONPath or Jinja2):

• JSONPath syntax (starting with $): Use for direct field extraction
  • Example: "output": "$.choices[0].message.content" extracts a nested field
  • Example: "session_id": "$.conv_id" extracts a top-level field
• Jinja2 templates: Use with jsonpath() function for conditional logic or complex extraction
  • Example: "output": "{{ jsonpath('$.text_response') or jsonpath('$.result.content') }}" tries the first path, falls back to second if empty
  • Pure JSONPath expressions (starting with $) work directly without Jinja2

For conditional logic or fallback values, use Jinja2 templates with jsonpath():

@endpoint(
    response_mapping={
        "output": "{{ jsonpath('$.text_response') or jsonpath('$.result.content') }}",
        "conversation_id": "$.conv_id",
    },
)
def chat(input: str) -> dict:
    """Extract output from multiple possible locations."""
    # Function may return different structures
    return {"text_response": "...", "conv_id": "abc123"}

Parameter Binding

The bind parameter allows you to inject infrastructure dependencies (database connections, configuration, auth context) into your endpoint functions without exposing them in the remote function signature.

Why Use Parameter Binding?

When your endpoint needs dependencies like database connections or authentication context, you have two options:

1. Framework DI (e.g., FastAPI’s Depends()): Use this for HTTP endpoints where the framework manages the request lifecycle
2. bind parameter: Use this for SDK endpoints that need dependencies in both local execution and remote testing

Basic Usage

from rhesis.sdk import endpoint

# Bind a static configuration object
@endpoint(
    bind={
        "config": AppConfig()  # Evaluated once at decoration time
    }
)
def process_request(config, input: str) -> dict:
    """config is injected automatically, only input appears in remote signature."""
    return {"output": f"Using {config.api_url}: {input}"}

Late Binding with Callables

Use callables (lambdas or functions) for dependencies that should be evaluated fresh on each call:

@endpoint(
    bind={
        "db": lambda: get_db_session(),  # Fresh connection per call
        "user": lambda: get_current_user_context(),  # Runtime context
    }
)
async def authenticated_query(db, user, input: str) -> dict:
    """db and user are injected, only input appears in remote signature."""
    if not user.is_authenticated:
        return {"output": "Unauthorized"}
    
    results = db.query_for_user(user.id, input)
    return {"output": format_results(results)}

Real-World Example

from rhesis.sdk import RhesisClient, endpoint
from myapp.database import get_db_with_tenant
from myapp.config import AppConfig
from myapp.auth import get_current_user

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

@endpoint(
    bind={
        "db": lambda: get_db_with_tenant(org_id="org_123"),
        "config": AppConfig(),
        "user": lambda: get_current_user(),
    }
)
async def query_mcp(db, config, user, input: str) -> dict:
    """
    All infrastructure dependencies are bound.
    Remote signature only shows: query_mcp(input: str)
    """
    # Validate permissions
    if not user.has_permission("query_mcp"):
        return {"output": "Permission denied"}
    
    # Use injected dependencies
    results = await db.query(
        table=config.mcp_table,
        query=input,
        user_id=user.id
    )
    
    return {
        "output": format_results(results),
        "session_id": user.session_id,
    }

Key Concepts

Excluded from Remote Signature

Bound parameters don’t appear in the registered function signature, so remote tests only need to provide business logic parameters:

# Function definition
@endpoint(bind={"db": lambda: get_db(), "config": AppConfig()})
def query_data(db, config, input: str, session_id: str = None):
    ...
 
# Remote signature (what tests see)
query_data(input: str, session_id: str = None)

Evaluation Timing

• Static values: Evaluated once at decoration time
• Callables: Evaluated fresh on each function call

config = AppConfig()  # Created once

@endpoint(
    bind={
        "config": config,  # Same instance every call
        "db": lambda: get_db(),  # Fresh connection every call
    }
)
def my_endpoint(config, db, input: str):
    ...

No Override

Bound parameters won’t override explicitly provided values:

@endpoint(bind={"db": lambda: get_default_db()})
def query(db, input: str):
    return {"output": db.query(input)}

# Explicitly provide db - uses provided value, not bound value
result = query(db=custom_db, input="test")

When to Use bind vs Framework DI

Use bind when:

• Building SDK endpoints for remote testing
• Need same dependencies in local execution and remote tests
• Working outside web framework context
• Want dependencies to work seamlessly with Rhesis platform

Use Framework DI (e.g., FastAPI Depends()) when:

• Building HTTP endpoints only
• Framework manages request lifecycle
• Need framework-specific features (request context, middleware)
• Not using Rhesis SDK connector

Can use both:

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

app = FastAPI()

# FastAPI endpoint with Depends
@app.post("/chat")
def fastapi_chat(
    input: str,
    db = Depends(get_db),  # FastAPI DI
):
    return process_chat(db, input)

# SDK endpoint with bind for remote testing
@endpoint(bind={"db": lambda: get_db()})  # SDK bind
def sdk_chat(db, input: str):
    return process_chat(db, input)

Examples

Multiple Functions

from rhesis.sdk import RhesisClient, collaborate

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

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

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

Custom Fields

@endpoint(
    request_mapping={
        "question": "{{ input }}",
        "policy_id": "{{ policy_number }}",
    },
    response_mapping={"output": "$.answer"},
)
def insurance_query(question: str, policy_id: str) -> dict:
    """Query insurance policy."""
    return {"answer": lookup_policy(question, policy_id)}

Platform Integration

Viewing Endpoints

Registered endpoints appear in the Rhesis dashboard:

• Location: Projects → Your Project → Endpoints
• Connection Type: SDK
• Status: Active (connected) or Inactive (disconnected)
• Naming: {Project Name} ({function_name})

Connection Management

• Reconnection: SDK automatically reconnects with exponential backoff if connection is lost
• Re-registration: Functions are automatically re-registered on reconnect
• Function changes: Adding/modifying functions updates endpoints automatically; removing functions marks endpoints as Inactive

Best Practices

• Use connectors when: Functions are in your codebase, using standard patterns, want code-first definition
• Use manual config when: Testing external APIs, need complex transformations, services outside codebase
• Function naming: Use descriptive names (avoid function1, handler)
• Type hints: Always include type hints for better auto-detection
• Error handling: Return structured error responses: {"output": None, "status": "error", "error": str(e)}
• Parameter binding: Use bind for infrastructure dependencies (db, config, auth) that shouldn’t appear in remote signatures
• Callable bindings: Use lambdas for dependencies that need fresh values per call (database connections, auth context)
• Static bindings: Use direct values for singletons (configuration objects, shared resources)
• Avoid binding business logic: Only bind infrastructure dependencies, not business logic parameters

Troubleshooting

Functions not appearing:

• Verify RhesisClient initialized with api_key, project_id, and environment
• Check functions use @endpoint() decorator
• Check logs for “Connector initialized” and “Sent registration” messages

Connection issues:

• Verify API key and project ID are correct
• Ensure Rhesis backend is accessible
• Check firewall/network settings for WebSocket connections

Mapping problems:

• Auto-mapping: Use standard field names (input, session_id, output)
• Manual mapping: Verify Jinja2/JSONPath syntax
• Custom fields: Ensure custom request fields are included in API requests

Common errors:

• "RhesisClient not initialized": Create RhesisClient instance before using @endpoint
• "@endpoint requires project_id": Provide project_id or set RHESIS_PROJECT_ID env var
• "Functions not appearing as Active": Restart app to trigger re-registration

Parameter binding issues:

• Bound parameter not injected: Ensure the parameter name in bind matches the function parameter name exactly
• Callable not evaluated: Check that you’re passing a callable (lambda or function), not calling it (e.g., lambda: get_db() not get_db())
• Static value stale: If using a static value that changes, use a callable instead: bind={"config": lambda: get_config()}
• Exception in bound callable: Exceptions in bound callables are propagated - ensure your dependency functions handle errors gracefully