> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.rubrichq.io/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.rubrichq.io/_mcp/server.

# Text Testing

Text testing runs RubricHQ's AI-simulated caller against your agent **over text instead of voice**. It's \~10× faster and far cheaper than voice testing — ideal for workflow validation, regression suites, and CI.

Text testing uses a **single format**: your agent exposes a WebSocket server speaking the **X-RUBRIC protocol**. Configure it under **Agent → Channels → Text Channel**. **RubricHQ is the client** — it connects out to *your* agent; your agent never connects to RubricHQ. When the conversation ends, RubricHQ records the transcript and evaluates your metrics automatically.

The X-RUBRIC WebSocket protocol gives you a secret handshake, scenario/run identification headers, end-of-conversation signaling, and **function-call visibility** in the transcript.

---

## Custom WebSocket

### What you set in RubricHQ (Agent → Channels → Text Channel)

These are values **you enter once** in the RubricHQ dashboard for the agent:

| Field                        | Description                                                                                                                                                                                     |
| ---------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **WebSocket URL**            | `wss://your-agent.com/ws` (must start with `wss://` in production)                                                                                                                              |
| **Secret**                   | A shared secret you choose. Store the same value in your server. RubricHQ sends it on every connection as the `X-RUBRIC-SECRET` header so your server can verify the caller is really RubricHQ. |
| **Per-turn timeout (s)**     | How long RubricHQ waits for a reply to each message before failing the run. Default `120`.                                                                                                      |
| **Max session duration (s)** | Hard cap on the whole conversation. Default `600`.                                                                                                                                              |

### What RubricHQ sends to your server on connect

RubricHQ includes **all** of these HTTP headers on **every** connection (the `X-…` row is the only conditional one — it appears only when the scenario has matching metadata). You **must validate `X-RUBRIC-SECRET`**; the others identify the agent / scenario / run so you can route and correlate.

| Header                 | Always sent | What it's for                                                                                                  |
| ---------------------- | ----------- | -------------------------------------------------------------------------------------------------------------- |
| `X-RUBRIC-SECRET`      | Yes         | Authenticate the connection — compare to your stored secret; reject with WS close `1008` if it doesn't match.  |
| `X-RUBRIC-AGENT-ID`    | Yes         | Which RubricHQ agent this conversation is for. Use it to route when one endpoint serves multiple bots.         |
| `X-RUBRIC-SCENARIO-ID` | Yes         | The scenario being tested.                                                                                     |
| `X-RUBRIC-RUN-ID`      | Yes         | This test run's id.                                                                                            |
| `X-RUBRIC-BATCH-ID`    | Yes         | The batch id (groups runs).                                                                                    |
| `X-…`                  | When set    | Any scenario metadata field whose name starts with `X-`, forwarded verbatim — your own custom routing/context. |

### Messages — RubricHQ → your agent

Standard user turn:

```json
{ "content": "I'd like to set up a payment plan" }
```

End signal (RubricHQ decided the conversation is over):

```json
{ "content": "Thanks, goodbye!", "type": "end_call" }
```

### Messages — your agent → RubricHQ

Standard agent reply:

```json
{ "content": "Sure — what monthly amount works for you?" }
```

Reply with metadata (merged into the run, last-write-wins; not a turn on its own):

```json
{ "content": "One moment…", "metadata": { "customer_id": "cust_123" } }
```

Metadata-only frame (recorded, **not** treated as a turn):

```json
{ "metadata": { "internal_call_id": "abc-123" } }
```

Function call + result (recorded in the transcript as `function_call` / `function_call_result`, **not** treated as a turn — keep streaming until you send the actual reply):

```json
{ "role": "Function Call", "data": { "id": "call_1", "name": "lookup_account", "arguments": "{\"id\":\"123\"}" } }
{ "role": "Function Call Result", "data": { "id": "call_1", "result": "{\"balance\":420.00}" } }
```

End the conversation (either form):

```json
{ "content": "Goodbye!", "type": "end_call" }
```

…or simply **close the socket** — RubricHQ treats an unexpected close as an agent-initiated end.

### Lifecycle & timeouts

* Either side may end via `{"type": "end_call"}`, or your server can close the socket.
* **Per-turn idle timeout** — if your agent doesn't reply within *Per-turn timeout*, the run ends `failed` (`ended_reason: agent_timeout`) with the partial transcript saved.
* **Max session duration** — the whole conversation is capped; exceeding it ends the run `failed` (`max_duration`).
* A failed connection ends the run `connection_failed`; a mid-conversation drop ends it `connection_lost`.

### A quick conversation (WebSocket)

A full session looks like this — each line is one frame; the label shows who sent it:

```text
   RubricHQ connects  (header  X-RUBRIC-SECRET: shh)
your agent →  {"content": "Hi, thanks for contacting Acme. How can I help?"}
RubricHQ   →  {"content": "I'd like to set up a payment plan"}
your agent →  {"content": "Sure — what monthly amount works for you?"}
RubricHQ   →  {"content": "About $150 a month"}
your agent →  {"content": "Got it. Can you tell me a little more?"}
RubricHQ   →  {"content": "Actually that's all, thanks!", "type": "end_call"}
   conversation ends → RubricHQ records the transcript and runs your metrics
```

A run with a function call looks the same, but the agent streams the call/result frames *before* its spoken reply (they're recorded in the transcript but don't count as a turn):

```text
RubricHQ   →  {"content": "What's my balance?"}
your agent →  {"role": "Function Call",        "data": {"id": "c1", "name": "lookup_account", "arguments": "{}"}}
your agent →  {"role": "Function Call Result", "data": {"id": "c1", "result": "{\"balance\":420.00}"}}
your agent →  {"content": "Your balance is $420.00."}
```

> **Tip**
>
> **Local development:** expose your local server with a tunnel (e.g. ngrok) and use the `wss://` URL in the WebSocket URL field.

### Your reply comes from *your* agent

Your agent's reply is produced by **your real agent** — an LLM, a Pipecat pipeline, a LangChain chain, your own model and tools. The WebSocket handler is only a thin bridge:

`receive {"content": user_text}` → **your agent (LLM + your context / tools / state)** → `send {"content": reply}`

So making an existing agent "WebSocket-ready" is just adding that bridge around your real brain. If your agent is built on **Pipecat**, you already have an LLM context and pipeline — feed the incoming text into your context, run the pipeline (text in → text out), and send the assistant's text back. A full Pipecat reference ships in the voice-pipecat project, `custom_ws_pipecat_agent.py`: a real `OpenAILLMContext` + `OpenAILLMService` pipeline bridged to this protocol (vs. `custom_ws_agent.py`, the minimal hand-rolled version).

### Example: a Pipecat agent

If your agent is a Pipecat pipeline, the bridge is small — push each incoming `content` into your LLM context, run the pipeline, and send the assistant's text back:

```python
# pip install websockets pipecat-ai openai
import asyncio, json, os, websockets
from pipecat.frames.frames import LLMTextFrame, LLMFullResponseEndFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineTask
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.services.openai.llm import OpenAILLMService

SECRET = os.environ["AGENT_SECRET"]
SYSTEM = "You are a helpful billing agent for Acme Corp. Be brief and professional."

class ResponseCapture(FrameProcessor):
    """Collect the assistant's streamed text; resolve one future per turn."""
    def __init__(self):
        super().__init__(); self._buf = ""; self._fut = None
    def expect(self):
        self._buf = ""; self._fut = asyncio.get_event_loop().create_future(); return self._fut
    async def process_frame(self, frame, direction):
        await super().process_frame(frame, direction)
        if isinstance(frame, LLMTextFrame):
            self._buf += frame.text or ""
        elif isinstance(frame, LLMFullResponseEndFrame) and self._fut and not self._fut.done():
            self._fut.set_result(self._buf.strip())
        await self.push_frame(frame, direction)

async def handler(ws):
    if ws.request.headers.get("X-RUBRIC-SECRET") != SECRET:
        await ws.close(1008, "Unauthorized"); return

    # Your existing Pipecat "brain": context + LLM + aggregator.
    context = OpenAILLMContext(messages=[{"role": "system", "content": SYSTEM}])
    llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o-mini")
    agg = llm.create_context_aggregator(context)        # appends assistant replies to context
    capture = ResponseCapture()
    task = PipelineTask(Pipeline([llm, capture, agg.assistant()]))
    runner_task = asyncio.create_task(PipelineRunner(handle_sigint=False).run(task))

    async def ask(text):                                 # one user turn -> assistant reply
        context.add_message({"role": "user", "content": text})
        fut = capture.expect()
        await task.queue_frame(OpenAILLMContextFrame(context))
        return await asyncio.wait_for(fut, timeout=30)

    try:
        await ws.send(json.dumps({"content": "Hi, thanks for contacting Acme. How can I help?"}))
        async for raw in ws:
            msg = json.loads(raw)
            if msg.get("type") == "end_call":
                break
            await ws.send(json.dumps({"content": await ask(msg.get("content", ""))}))
    finally:
        runner_task.cancel()

async def main():
    async with websockets.serve(handler, "0.0.0.0", 8080):
        await asyncio.Future()

asyncio.run(main())
```

That's the heart of the full reference `custom_ws_pipecat_agent.py`, which additionally routes on `X-RUBRIC-AGENT-ID` and parses inline `<dtmf/>` tags. Any other framework (LangChain, your own model) follows the same shape: `content` in → your agent → `content` out.

---

## DTMF (keypad input)

When a scenario enables the `send_dtmf` tool, the simulated caller can "press keypad digits" during a text conversation. Rather than a separate frame, it sends the digits as an **inline tag embedded in a normal text message**:

```json
{ "content": "Sure, entering it now. <dtmf digits=\"1234#\"/>" }
```

Your agent should parse `<dtmf digits=\"...\"/>` out of the message `content`, treat the matched digits as the caller's keypad entry, and strip the tag before handling the text. Valid characters are `0-9`, `*`, and `#`. A simple regex works:

```python
import re
m = re.search(r'<dtmf\s+digits="([^"]*)"\s*/>', content)
digits = m.group(1) if m else None
```

The full message (tag included) is recorded in the transcript, so DTMF entries are visible in the run results.

---

## Testing multiple agents on one endpoint

In RubricHQ you create **one agent per bot** you want to test. If those bots share a **single endpoint**, your server still needs to know *which* bot a given request/connection is for. RubricHQ tells you in two ways — pick whichever fits:

**Option A — Read the `X-RUBRIC-AGENT-ID` header.** Every WebSocket connection carries `X-RUBRIC-AGENT-ID`, the RubricHQ agent id. Map each agent id to your bot once (the id is stable per agent), then dispatch on the header:

```python
agent_id = headers.get("X-RUBRIC-AGENT-ID")   # e.g. "7"
bot = MY_BOTS[agent_id]                         # your mapping
```

**Option B — Put the bot in the URL.** Give each RubricHQ agent a distinct **path or query** as its WebSocket URL — works on the same host and port:

```
wss://host/ws/billing      (or  wss://host/ws?bot=billing)
```

RubricHQ connects to exactly the URL configured for that agent, so your router reads the path/query and dispatches.

You can also attach **your own** identifier: any scenario metadata field whose name starts with `X-` is forwarded verbatim as a header, so e.g. a scenario field `X-Bot-Id = billing` arrives as the `X-Bot-Id` header. (Use this for per-scenario routing; use `X-RUBRIC-AGENT-ID` or the URL for per-agent routing.)

> **Tip**
>
> Validate `X-RUBRIC-SECRET` first, then route on `X-RUBRIC-AGENT-ID` (or the URL). The other headers (`X-RUBRIC-SCENARIO-ID`, `X-RUBRIC-RUN-ID`, `X-RUBRIC-BATCH-ID`) identify the specific scenario / run / batch within that agent.

## Test your endpoint before going live

You don't need to launch a full RubricHQ run to validate your endpoint. Three levels of dev testing:

**1. Probe it yourself (no RubricHQ needed).** This small script mimics exactly what RubricHQ sends — it connects with the `X-RUBRIC-*` headers, sends a few user turns, and prints every frame your agent returns. Point it at your `wss://` URL (local or staging):

```python
# rubric_ws_probe.py   —   pip install websockets
#   python rubric_ws_probe.py wss://localhost:8080/ws  your-secret
import asyncio, json, sys, websockets

async def main(url, secret):
    headers = {
        "X-RUBRIC-SECRET": secret,
        "X-RUBRIC-AGENT-ID": "probe",
        "X-RUBRIC-SCENARIO-ID": "1",
        "X-RUBRIC-RUN-ID": "1",
        "X-RUBRIC-BATCH-ID": "1",
    }
    async with websockets.connect(url, additional_headers=headers) as ws:
        print("connected:", url)
        for turn in ["Hi, I'd like to check my balance", "My account is 12345", "No, that's all — thanks"]:
            await ws.send(json.dumps({"content": turn})); print("→", turn)
            while True:                         # your agent may send fn-call / metadata before its reply
                msg = json.loads(await asyncio.wait_for(ws.recv(), timeout=30)); print("←  ", msg)
                if msg.get("type") == "end_call": return
                if "content" in msg: break
        await ws.send(json.dumps({"content": "", "type": "end_call"})); print("→ end_call")

asyncio.run(main(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else ""))
```

**2. Dry-run one scenario in the app.** Use **Live Simulations** to run a *single* scenario against your endpoint and watch the transcript and metrics live, instead of launching a full batch — ideal for the first real round-trip.

**3. Local development.** Expose your local server with a tunnel (e.g. `ngrok http 8080`) so RubricHQ — or the probe above — can reach `wss://…` while you iterate.

Once the probe and a single Live Simulation pass, you're ready to run full batches in production.

## Metrics on text runs

* **Transcript / LLM metrics** (compliance, resolution, custom LLM judges, etc.) run on text exactly as on voice.
* **Audio metrics** (latency, interruptions, WPM, voice clarity…) are **automatically skipped** for text — there's no audio.
* When creating a custom metric, **Applies to** lets you choose **Text and Voice conversations** (default) or **Text Conversation only**. A text-only metric won't run on voice; anything that runs on voice also runs on text.

## Results

After a run completes you'll see the **transcript** (including any function-call frames), the **metric evaluations**, and the **end reason** if the run failed (e.g. *Agent did not respond*). Text runs show no recording section.