LivePython platform

A platform for voice and chat agents

Problem

I wanted agents for phone, chat and messenger that are set up through configuration instead of code. They should also be able to run on a self-hosted server and only do what they are allowed to do.

My role

I designed and built the platform: specification, architecture, security model, conversation flows and prompts, tests, deployment on the VPS and the configuration of each agent. Part of the code came from cloud routines that open pull requests and have a second role review them. Only I deploy.

Approach

An agent is a folder under agents/<name>/: agent.yaml for persona, models and knowledge, flow.yaml as a conversation graph with the allowed actions per phase, actions.yaml for follow-ups such as email, calendar or callback, plus a knowledge folder. No custom code per agent. The pipeline is built on Pipecat 1.5 and FastAPI. STT, LLM and TTS come from a provider factory, local or cloud, with automatic failover to a backup model at a different provider. Knowledge is searched by a hybrid RAG in LanceDB: full text with a German tokenizer plus vector search, merged with RRF, using local embeddings.

For AI assistants there is a dedicated MCP server in its own container: OAuth 2.1 with its own login and TOTP, permissions as scopes, and changing platform actions only inside a write window that I open in the cockpit. The one exception is narrow: an email to myself. I secure quality with a pytest suite and golden transcripts that run real flows with fixed model answers. The platform is in use as the web chat of Seniorenbetreuung Lala in Düsseldorf and as the Telegram bot of Haus Wessels / Hotel Nike in Bottrop.

Architecture

Channels
Web chat widget, voice via WebRTC in the browser, Telegram as a back channel per agent, with its own bot. Phone network via Telnyx built, not live yet.
FastAPI and Pipecat
STT, LLM and TTS from the provider factory, LLM failover via the LLMSwitcher to a second provider.
Agent as configuration
agents/<name>/ with agent.yaml, flow.yaml (conversation graph, allowed actions per phase), actions.yaml and knowledge/.
Hybrid RAG
LanceDB with full text (German tokenizer) and vectors, RRF, local embeddings via fastembed, fallback to vector-only search.
Actions and log
Calendar, email, webhook, callback. A failing action never ends the conversation. Every conversation in SQLite with transcript, actions and cost.
Own MCP server
Separate container. OAuth 2.1, login with TOTP, scopes, write window set from the cockpit for changing platform actions (exception: an email to myself), log per tool call. No shell, no exec.
Quality
pytest suite, golden transcripts with fixed model answers, tests on every push to main.

Decision

Fail fast instead of silent retries

On the phone, silence costs more than a weaker model. The OpenAI SDK silently retries on 429 and 5xx and waits out the Retry-After. Groq sends 20-60 seconds when the daily limit is hit, which means minutes of silence in a call. I turn off the SDK retries, wait out short limits of up to 8 seconds exactly once, and pass everything else straight to Pipecat's LLMSwitcher, which moves to a model at a different provider. I ignore late errors from the service that has already been replaced. Otherwise the switcher flips back to the broken model, which I saw in the logs: a switch, and 72 ms later back again.

Code excerpt

core/providers.py (shortened)Scroll sidewayspython

def _fail_fast(service_cls):
    """OpenAI-kompatible Services ohne stille SDK-Retries: 429/5xx schlafen
    sonst den Retry-After ab (Groq schickt beim Tages-Limit 20-60 s!) - am
    Telefon ist das Minuten-Stille. Kurze Minuten-Limits (s.o.)
    wartet der Transport einmal ab; alles andere reicht sofort hoch, der
    LLMSwitcher failovert in Millisekunden aufs Fallback-Modell."""

    class FailFast(service_cls):
        def create_client(self, *args, **kwargs):
            client = super().create_client(*args, **kwargs).with_options(max_retries=0)
            # ...
            return client
    # ...

class _FailoverIgnoringStaleErrors(ServiceSwitcherStrategyFailover):
    # ...
    async def handle_error(self, error):
        failed = getattr(error, "processor", None)
        if failed is not None and failed is not self._active_service:
            logger.debug(f"Failover: ignoriere verspäteten Fehler von {failed.name}")
            return None
        self.letzter_fehler = str(getattr(error, "error", "") or "")
        return await super().handle_error(error)

SDK retries off, so the LLMSwitcher moves to the backup model right away. Errors from a service that has already been replaced are ignored, otherwise the switcher swings back.

Results

7330tests passing in the full pytest suite before the deploy on 18 Oct 2026
8golden transcript scenarios that run real flows with fixed model answers
2live deployments: web chat at Seniorenbetreuung Lala, Telegram bot at Haus Wessels / Hotel Nike
1256commits since 7 Jul 2026

What I learned

I built too wide. On 23 Sep 2026, 44 work packages were built in three waves, but none of the phone components has been accepted with real telephony. Today I would first put the path into the phone network live and accept it in one real deployment, then expand. The components that proved themselves at Haus Wessels are the ones most tied to that single project. I would cut them generically earlier. And I would split server.py sooner, the file is now about 4,800 lines.