> ## Documentation Index
> Fetch the complete documentation index at: https://docs.coval.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Coval Agent SDK

> Connect an existing Python voice agent to Coval simulations through an outbound audio connection.

Connect your Python voice pipeline to Coval's simulated callers with `coval-agent-sdk`. Your application runs the agent; Coval records and evaluates the conversation. The SDK opens an outbound WebSocket, so your agent does not need a public endpoint.

## Install

Requires **Python 3.10 or newer**:

```sh theme={null}
python -m pip install coval-agent-sdk
```

Import the package as `coval_agent_sdk`. See [PyPI](https://pypi.org/project/coval-agent-sdk/) for versions and release status. To manage resources through the API, use the separate [API SDKs](/sdks/api-sdks.md).

## Before you run

Select these resources from the same Coval workspace:

* An organization API key with access to launch simulations and read their results.
* The workspace ID.
* An agent configured for the **SDK WebSocket** connection type.
* A voice persona and a test set containing your scenarios.

For a first run, one case and one iteration are enough.

You can also [manage test configuration with Terraform](/sdks/terraform-provider.md#use-it-with-the-agent-sdk).

## Check the connection

Pass your API key and workspace ID to `CovalVoiceClient`, and the agent, persona, and test-set IDs to `RunConfig`. This example reads those values from environment variables.

The handler below echoes the caller's audio to check the connection. In an agent integration, the handler connects that audio stream to your voice pipeline.

```python theme={null}
import asyncio
import os

import coval_agent_sdk


async def on_connection(connection: coval_agent_sdk.AudioConnection) -> None:
    async for audio in connection.receive_audio():
        await connection.send_audio(audio)


async def main() -> coval_agent_sdk.RunResult:
    client = coval_agent_sdk.CovalVoiceClient(
        api_key=os.environ["COVAL_API_KEY"],
        workspace_id=os.environ["COVAL_WORKSPACE_ID"],
    )
    config = coval_agent_sdk.RunConfig(
        agent_id=os.environ["COVAL_AGENT_ID"],
        persona_id=os.environ["COVAL_PERSONA_ID"],
        test_set_id=os.environ["COVAL_TEST_SET_ID"],
        run_timeout_seconds=600,
    )
    return await client.run(config, on_connection=on_connection)


if __name__ == "__main__":
    result = asyncio.run(main())
    print(result.run_id, result.status)
```

The client defaults to `https://api.coval.dev/v1`. Set `api_base_url` to use another API root, with credentials and resource IDs from that environment. `agent_id` identifies the agent configuration in Coval.

## Connect your voice pipeline

Pass an async handler to `run(config, on_connection=...)`. The SDK calls it with a ready `AudioConnection` for each conversation. The handler bridges your pipeline and the connection:

* Consume caller audio through `receive_audio()` and send agent speech with awaited `send_audio()` calls.
* Keep input and output independent so the agent can speak before receiving caller audio.
* Keep the handler active through voice completion, then finish conversation tasks and spans before returning.

For Pipecat, connect the receive loop to your input processor and send pipeline speech through your output processor. Your application handles resampling and provider setup.

Returning before voice completion raises an SDK error. A transport failure cancels and awaits the handler so its `finally` blocks can finish. Handlers may run concurrently; keep conversation state separate.

### Audio connection reference

| Member | Behavior |
| - | - |
| `simulation_id` | Identifies this conversation for application state and traces. |
| `receive_audio()` | Async iterator of caller PCM. Use one active iterator and drain it promptly. Normal completion ends the iterator; failures raise. |
| `send_audio(pcm)` | Await a paced output write. Returns `True` when fully written, or `False` if normal voice completion interrupts the write. Failures raise. |
| `wait_closed()` | Waits for voice completion without consuming input; failures raise. |
| `closed` | Reports whether the exchange ended, including failure. |
| `result` | Finalized `ConversationResult` after a successful connection context exits; otherwise `None`. |

### Audio format and limits

Both directions use **16 kHz, 16-bit (2 bytes per sample), mono, little-endian PCM**. A 20 ms frame contains 640 bytes. Convert your pipeline's input and output formats at its transport boundary; the SDK does not resample.

Await every send. Each chunk must be even-length `bytes`, at most 2 MiB. The SDK slices and paces output across calls and serializes concurrent writes. A write interrupted by normal completion returns `False`, even if some audio was sent.

Unread input is limited to 64 KiB; overflow fails the connection. Drain input promptly and bound application queues. The SDK does not send relay clear or barge-in controls.

## Launch a run or connect to one conversation

`run()` creates a run, invokes your handler for each ready connection, and waits for completion, including evaluation processing. `RunConfig` defaults to one iteration, concurrency one, and a 3,600-second deadline; the example uses 600 seconds.

`client.connect(simulation_id)` attaches to **one existing simulation** without launching a run. Its async context yields a ready connection:

```python theme={null}
async def attach(
    client: coval_agent_sdk.CovalVoiceClient, simulation_id: str
) -> coval_agent_sdk.ConversationResult | None:
    async with client.connect(simulation_id) as connection:
        await on_connection(connection)
    return connection.result
```

Read `connection.result` after the context exits successfully. Each connection has its own simulation ID; multiple iterations of a test case produce separate conversations.

Local timeout or cancellation stops local work without cancelling the hosted run. Inspect that run before retrying. The SDK does not automatically repeat run creation or reconnect a consumed connection.

## Results and errors

`run()` returns an immutable `RunResult` with:

* `run_id`: the hosted run identifier.
* `status`: the hosted completion status.
* `conversations`: a tuple of completed `ConversationResult` values.

Each conversation result contains `simulation_id`, local voice `status`, `received_audio_bytes`, `sent_audio_bytes`, and optional `trace` delivery information. With tracing disabled, `trace` is `None`.

A run result reports execution completion; a conversation result reports voice completion. Review evaluation results in Coval to check whether the agent met its quality criteria.

Catch `CovalError` for SDK failures. After launch, `RunError.result` retains the run ID, completed conversations, and last observed status. Application exceptions and cancellation propagate unchanged.

To recover the run ID after a failure, use this pattern inside your async application code:

```python theme={null}
try:
    result = await client.run(config, on_connection=on_connection)
except coval_agent_sdk.RunError as error:
    print("Inspect this run before retrying:", error.result.run_id)
    raise
```

### Argument validation

Missing arguments and unknown keywords raise `TypeError`. Invalid `RunConfig` limits or empty IDs raise `ValueError`. Credentials and resource access are checked when requests reach the API. See the [API contract](https://github.com/coval-ai/coval-agent-sdk/blob/main/docs/developer-api.md#argument-validation) for details.

## Optional OpenTelemetry traces

Install the extra:

```sh theme={null}
python -m pip install "coval-agent-sdk[tracing]"
```

Register a `CovalTraceExporter` with your OpenTelemetry provider and pass it to the client's `trace_exporter` argument. This example creates a provider and a handler with a conversation span:

```python theme={null}
from opentelemetry import trace
from opentelemetry.sdk import trace as sdk_trace
from opentelemetry.sdk.trace import export

from coval_agent_sdk import tracing

exporter = tracing.CovalTraceExporter()
provider = sdk_trace.TracerProvider()
provider.add_span_processor(export.SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("voice-agent")


async def traced_connection(
    connection: coval_agent_sdk.AudioConnection,
) -> None:
    with tracer.start_as_current_span(
        "conversation",
        attributes={"conversation.id": connection.simulation_id},
    ):
        await on_connection(connection)
```

Use your application's existing OpenTelemetry provider when one is configured. The client and provider share the same exporter. Conversation spans need a `conversation` root with `conversation.id` set to the simulation ID. Finish child spans before the handler returns, and shut down tracing after SDK cleanup.

The SDK uploads traces for each completed conversation. Check `ConversationResult.trace.uploaded_spans` and `trace.error` for delivery results. Trace delivery failures do not rerun the conversation.

Spans may contain prompts, transcripts, and model output. Export only data you intend to send to Coval. See the [trace lifecycle](https://github.com/coval-ai/coval-agent-sdk/blob/main/docs/runtime.md#trace-lifecycle) for buffering and cleanup details.

## Source and support

See the [SDK repository](https://github.com/coval-ai/coval-agent-sdk) for the implementation and API contracts. For integration help, contact [Coval support](mailto:support@coval.dev).


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.