Metadata-Version: 2.5
Name: callharness-sdk
Version: 0.1.1
Summary: Python SDK for CallHarness - open-source call analytics for voice AI agents
Project-URL: Homepage, https://github.com/mohammad-mussab/callharness
Project-URL: Repository, https://github.com/mohammad-mussab/callharness
Project-URL: Issues, https://github.com/mohammad-mussab/callharness/issues
Author: Mohammad Mussab
License: MIT
License-File: LICENSE
Keywords: call-analytics,evaluation,livekit,llm,observability,pipecat,voice-ai
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Multimedia :: Sound/Audio :: Speech
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.25
Provides-Extra: pipecat
Requires-Dist: pipecat-ai>=0.0.60; extra == 'pipecat'
Description-Content-Type: text/markdown

# callharness-sdk

Python SDK for [CallHarness](https://github.com/mohammad-mussab/callharness) — open-source call analytics for voice AI agents.

Send your agent's calls to a CallHarness server, which runs post-call LLM analysis
(summary, sentiment, outcome, why a call transferred or didn't complete) and shows it
on a dashboard. Self-hosted, so your transcripts stay on your own infrastructure.

## Install

```bash
pip install callharness-sdk            # REST client + turn assembly
pip install "callharness-sdk[pipecat]" # also installs pipecat-ai for the observers
```

The `[pipecat]` extra only adds `pipecat-ai`. Skip it if you're on LiveKit, a custom
stack, or calling the REST API directly — `callharness_sdk.pipecat` still imports
cleanly without it, and only raises if you actually instantiate an observer.

## Direct ingestion

```python
from callharness_sdk import CallHarnessClient

client = CallHarnessClient("http://localhost:8010")
call = client.ingest_call(
    agent_id="my-agent",
    end_reason="completed",
    turns=[
        {"role": "assistant", "text": "Hi, how can I help?"},
        {"role": "user", "text": "I'd like to book an appointment."},
    ],
)
client.upload_recording(call["id"], "recording.wav")
```

## Pipecat integration

```python
from pipecat.processors.transcript_processor import TranscriptProcessor
from callharness_sdk.pipecat import create_recorder

transcript = TranscriptProcessor()
recorder = create_recorder("http://localhost:8010", agent_id="my-agent")
recorder.attach(transcript)

# include transcript.user() after STT and transcript.assistant() after TTS
# in your pipeline, then when the call ends:
await recorder.flush(end_reason="completed")
```

To capture per-turn STT/LLM/TTS latency, add the metrics observer to your task:

```python
from callharness_sdk.pipecat import CallHarnessMetricsObserver

task = PipelineTask(
    pipeline,
    params=PipelineParams(enable_metrics=True),
    observers=[CallHarnessMetricsObserver(recorder)],
)
```

## Pipecat without TranscriptProcessor

If your pipeline doesn't use `TranscriptProcessor` (e.g. you capture transcripts at
the frame level), use the all-in-one frame observer instead — it captures transcript
turns, end-to-end response latency, STT/LLM/TTS components, interruptions, tool calls,
transfers, and a deterministic `end_reason`, all from one observer:

```python
from callharness_sdk.pipecat import CallHarnessFrameObserver, create_recorder

recorder = create_recorder("http://localhost:8010", agent_id="my-agent")
observer = CallHarnessFrameObserver(
    recorder, stt=stt, tts=tts, transfer_tool_names={"transfer_to_human"}
)
# add `observer` to your PipelineTask/PipelineWorker observers, then on call end:
await recorder.flush(
    end_reason=observer.finalize_end_reason(),  # "completed" | "transferred" | "error" | ...
    transferred=observer.transferred,
    recording_bytes=wav_bytes,   # optional in-memory recording upload
)
```

`finalize_end_reason()` gives you the best reason available *right now*: `"error"`
if a fatal `ErrorFrame` occurred, otherwise the explicit `reason=` you passed to
`EndTaskFrame`/`CancelTaskFrame` if the pipeline already saw one before you called
this (e.g. `EndTaskFrame(reason="silence_timeout")` from a `UserIdleProcessor`
callback), otherwise `"transferred"` if a transfer fired, otherwise `"completed"`
(pass a different `default=` if that's not right for your integration).

Call `finalize_end_reason()` — not the raw `observer.end_reason` attribute — from
your own disconnect/teardown handler (e.g. a transport's `on_client_disconnected`).
That fires *before* an `EndFrame`/`CancelFrame` has necessarily propagated through
the pipeline, so `.end_reason` may still be `None` at that point; `finalize_end_reason()`
only depends on state (fatal error, transferred) that's already known live during the
call, so it's correct regardless of teardown ordering. `observer.last_error` holds the
most recent error message, useful to drop into `metadata` for debugging.

## If your agent already has its own call pipeline

Mature agents usually already collect a transcript, their own record of tool calls,
and write to their own database. Adopting `CallRecorder` would mean maintaining two
sources of truth — so instead, hand CallHarness what you already have:

```python
from callharness_sdk import CallHarnessClient, LatencyCollector, assemble_turns
from callharness_sdk.pipecat import CallHarnessMetricsObserver

latency = LatencyCollector()          # satisfies what the observer expects
task = PipelineTask(
    pipeline,
    params=PipelineParams(enable_metrics=True),   # required, or no metrics are emitted
    observers=[CallHarnessMetricsObserver(latency)],
)

# ...at the end of the call, from your own save routine:
turns = assemble_turns(
    transcript=my_transcript,        # [{role, content, timestamp}, ...]
    tool_calls=my_function_calls,    # [{function_name, parameters, result, timestamp}]
    latency=latency,
    started_at=call_started_at,
)
CallHarnessClient("http://localhost:8010").ingest_call(
    agent_id="my-agent", turns=turns, external_id=my_call_id,
    transferred=..., metadata={"my_own_verdict": ...},
)
```

`assemble_turns()` does the fiddly part: a tool call and a latency sample both happen
*while* a reply is being produced, before its text exists, so both are matched by
timestamp to the assistant turn they actually belong to. It accepts either field
naming (`name`/`function_name`, `arguments`/`parameters`, `content`/`text`), truncates
oversized tool results, and never records a tool as successful unless it can prove it.

Anything you send in `metadata` is stored alongside the call — useful if your agent
already classifies its own calls and you want to compare that against CallHarness's
independent verdict.

## Full example

See [examples/pipecat_bot.py](https://github.com/mohammad-mussab/callharness/blob/main/examples/pipecat_bot.py)
for a complete working bot.

## Licence

MIT
