Coval Agent SDK - Coval Documentation

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:

python -m pip install coval-agent-sdk

Import the package as coval_agent_sdk. See PyPI for versions and release status. To manage resources through the API, use the separate API SDKs.

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.

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.

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:

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:

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 for details.

Optional OpenTelemetry traces

Install the extra:

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:

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 for buffering and cleanup details.

Source and support

See the SDK repository for the implementation and API contracts. For integration help, contact Coval support.

​Install

​Before you run

​Check the connection

​Connect your voice pipeline

​Audio connection reference

​Audio format and limits

​Launch a run or connect to one conversation

​Results and errors

​Argument validation

​Optional OpenTelemetry traces

​Source and support