Clawdbot Voice Chat — Pipecat + Deepgram + ElevenLabs (complete files)

.env.example

# Deepgram — get a free key at https://console.deepgram.com (200hrs free)
DEEPGRAM_API_KEY=your_deepgram_api_key

# ElevenLabs — your existing key
ELEVENLABS_API_KEY=your_elevenlabs_api_key

# ElevenLabs voice ID (default: "Josh" — a clear male voice)
# Browse voices at https://elevenlabs.io/voice-library
ELEVENLABS_VOICE_ID=TxGEqnHWrfWFTfGW9XjX

# Clawdbot gateway (defaults should work if running locally)
CLAWDBOT_GATEWAY_URL=http://127.0.0.1:18789
CLAWDBOT_GATEWAY_TOKEN=clawdbot-gateway-2026

.gitignore

.venv/
__pycache__/
*.pyc
.env

bot.py

"""Felix Voice Chat — real-time voice conversation with Felix via Pipecat + Clawdbot."""

import os

from dotenv import load_dotenv
from loguru import logger

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.frames.frames import LLMRunFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.elevenlabs.tts import ElevenLabsTTSService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport

load_dotenv(override=True)

# Voice-specific system instruction — the gateway injects the full Felix context
# (SOUL.md, memory, tools, etc.) on top of this.
VOICE_SYSTEM = (
    "This conversation is happening via real-time voice chat. "
    "Keep responses concise and conversational — a few sentences at most "
    "unless the topic genuinely needs depth. "
    "No markdown, bullet points, code blocks, or special formatting. "
    "Don't use emoji. Speak naturally as Felix would in conversation."
)


async def run_bot(webrtc_connection):
    # --- STT: Deepgram streaming ---
    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))

    # --- TTS: ElevenLabs ---
    tts = ElevenLabsTTSService(
        api_key=os.getenv("ELEVENLABS_API_KEY"),
        voice_id=os.getenv("ELEVENLABS_VOICE_ID", "TxGEqnHWrfWFTfGW9XjX"),
    )

    # --- LLM: Clawdbot gateway (OpenAI-compatible endpoint) ---
    gateway_url = os.getenv("CLAWDBOT_GATEWAY_URL", "http://127.0.0.1:18789")
    llm = OpenAILLMService(
        api_key=os.getenv("CLAWDBOT_GATEWAY_TOKEN", "clawdbot-gateway-2026"),
        model="clawdbot:voice",
        base_url=f"{gateway_url}/v1",
    )

    # --- Conversation context ---
    messages = [
        {"role": "system", "content": VOICE_SYSTEM},
    ]

    context = LLMContext(messages)
    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(context)

    # --- Transport: self-hosted WebRTC ---
    transport = SmallWebRTCTransport(
        webrtc_connection=webrtc_connection,
        params=TransportParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
            vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.4)),
            audio_out_10ms_chunks=2,
        ),
    )

    # --- Pipeline ---
    pipeline = Pipeline(
        [
            transport.input(),       # Browser audio in
            stt,                     # Speech → text (Deepgram)
            user_aggregator,         # Accumulate user turns
            llm,                     # Felix via Clawdbot gateway
            tts,                     # Text → speech (ElevenLabs)
            transport.output(),      # Audio back to browser
            assistant_aggregator,    # Track assistant turns
        ]
    )

    task = PipelineTask(
        pipeline,
        params=PipelineParams(
            enable_metrics=True,
            enable_usage_metrics=True,
        ),
    )

    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
        logger.info("Client connected — starting conversation")
        messages.append(
            {
                "role": "system",
                "content": "Nat just connected to voice chat. Say a brief hello.",
            }
        )
        await task.queue_frames([LLMRunFrame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        logger.info("Client disconnected")
        await task.cancel()

    runner = PipelineRunner(handle_sigint=False)
    await runner.run(task)

gist-readme.md

Clawdbot Voice Chat — Complete Files

Everything you need to add real-time voice chat to your Clawdbot setup. Uses Pipecat + Deepgram STT + ElevenLabs TTS, routed through Clawdbot's gateway so you get the full agent (memory, tools, persona).

Read the full writeup: (link to article)

Quick Start

# 1. Install uv if you don't have it
curl -LsSf https://astral.sh/uv/install.sh | sh

# 2. Create the project
mkdir -p ~/clawd/voice-chat
cd ~/clawd/voice-chat
# Download or paste the files from this gist

# 3. Configure
cp .env.example .env
# Edit .env with your Deepgram + ElevenLabs keys
# Update CLAWDBOT_GATEWAY_TOKEN to match your gateway auth token

# 4. Add voice agent to clawdbot.json (if you don't have one)
# Under agents.list, add:
#   { "id": "voice", "workspace": "/path/to/your/clawd", "model": "anthropic/claude-sonnet-4-5" }

# 5. Make sure chatCompletions is enabled in clawdbot.json:
# gateway.http.endpoints.chatCompletions.enabled = true

# 6. Run
uv run server.py

# 7. Open http://localhost:7860 and click the mic

Clawdbot Config

Two things needed in clawdbot.json:

Enable chat completions endpoint:

{
  "gateway": {
    "http": {
      "endpoints": {
        "chatCompletions": { "enabled": true }
      }
    }
  }
}

Add a voice agent:

{
  "agents": {
    "list": [
      {
        "id": "voice",
        "workspace": "/path/to/your/clawd",
        "model": "anthropic/claude-sonnet-4-5"
      }
    ]
  }
}

The model field in bot.py is "clawdbot:voice" — the part after the colon maps to this agent id. Sonnet is a good default for voice latency. Opus gives better reasoning but adds a noticeable pause.

index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Felix Voice Chat</title>
    <style>
        * { margin: 0; padding: 0; box-sizing: border-box; }
        body {
            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
            background: #1a1a2e;
            color: #eee;
            height: 100vh;
            display: flex;
            flex-direction: column;
            align-items: center;
            justify-content: center;
        }
        .container {
            text-align: center;
            max-width: 400px;
        }
        .avatar {
            font-size: 64px;
            margin-bottom: 16px;
        }
        h1 {
            font-size: 28px;
            font-weight: 600;
            margin-bottom: 8px;
        }
        .subtitle {
            color: #888;
            font-size: 14px;
            margin-bottom: 40px;
        }
        #status {
            font-size: 15px;
            margin-bottom: 32px;
            min-height: 24px;
            color: #aaa;
        }
        #status.connected { color: #4ade80; }
        #status.connecting { color: #facc15; }
        #status.error { color: #f87171; }

        #connect-btn {
            width: 80px;
            height: 80px;
            border-radius: 50%;
            border: none;
            background: #3b82f6;
            color: white;
            font-size: 28px;
            cursor: pointer;
            transition: all 0.2s;
            display: flex;
            align-items: center;
            justify-content: center;
            margin: 0 auto;
        }
        #connect-btn:hover { background: #2563eb; transform: scale(1.05); }
        #connect-btn.active {
            background: #ef4444;
            animation: pulse 2s infinite;
        }
        #connect-btn.active:hover { background: #dc2626; }

        @keyframes pulse {
            0%, 100% { box-shadow: 0 0 0 0 rgba(239, 68, 68, 0.4); }
            50% { box-shadow: 0 0 0 12px rgba(239, 68, 68, 0); }
        }

        .hint {
            margin-top: 32px;
            font-size: 12px;
            color: #555;
        }
    </style>
</head>
<body>
    <div class="container">
        <div class="avatar">📋</div>
        <h1>Felix</h1>
        <p class="subtitle">Voice Chat</p>
        <p id="status">Ready</p>
        <button id="connect-btn">🎙️</button>
        <p class="hint">Click to connect · Allow microphone access</p>
    </div>

    <audio id="audio-el" autoplay></audio>

    <script>
        const statusEl = document.getElementById("status")
        const buttonEl = document.getElementById("connect-btn")
        const audioEl = document.getElementById("audio-el")

        let connected = false
        let peerConnection = null

        const sendIceCandidate = async (pc, candidate) => {
            await fetch('/api/offer', {
                method: "PATCH",
                headers: { "Content-Type": "application/json" },
                body: JSON.stringify({
                    pc_id: pc.pc_id,
                    candidates: [{
                        candidate: candidate.candidate,
                        sdp_mid: candidate.sdpMid,
                        sdp_mline_index: candidate.sdpMLineIndex
                    }]
                })
            });
        };

        const connect = async () => {
            setStatus("Connecting…", "connecting")
            buttonEl.classList.add("active")
            buttonEl.textContent = "⏹️"
            connected = true

            try {
                const audioStream = await navigator.mediaDevices.getUserMedia({ audio: true })
                const config = {
                    iceServers: [{ urls: "stun:stun.l.google.com:19302" }]
                };
                const pc = new RTCPeerConnection(config)
                peerConnection = pc
                pc.pendingIceCandidates = []
                pc.canSendIceCandidates = false

                pc.ontrack = e => audioEl.srcObject = e.streams[0]
                pc.addTransceiver(audioStream.getAudioTracks()[0], { direction: 'sendrecv' })
                pc.addTransceiver('video', { direction: 'sendrecv' })

                pc.onconnectionstatechange = () => {
                    if (pc.connectionState === 'connected') {
                        setStatus("Connected — speak anytime", "connected")
                    } else if (pc.connectionState === 'disconnected' || pc.connectionState === 'failed') {
                        disconnect()
                    }
                }

                pc.onicecandidate = async (event) => {
                    if (event.candidate) {
                        if (pc.canSendIceCandidates && pc.pc_id) {
                            await sendIceCandidate(pc, event.candidate)
                        } else {
                            pc.pendingIceCandidates.push(event.candidate)
                        }
                    }
                };

                await pc.setLocalDescription(await pc.createOffer())
                const response = await fetch('/api/offer', {
                    body: JSON.stringify({ sdp: pc.localDescription.sdp, type: pc.localDescription.type }),
                    headers: { 'Content-Type': 'application/json' },
                    method: 'POST',
                });
                const answer = await response.json()
                pc.pc_id = answer.pc_id
                await pc.setRemoteDescription(answer)

                pc.canSendIceCandidates = true
                for (const candidate of pc.pendingIceCandidates) {
                    await sendIceCandidate(pc, candidate)
                }
                pc.pendingIceCandidates = []
            } catch (err) {
                console.error(err)
                setStatus("Error: " + err.message, "error")
                disconnect()
            }
        }

        const disconnect = () => {
            if (peerConnection) {
                peerConnection.close()
                peerConnection = null
            }
            connected = false
            buttonEl.classList.remove("active")
            buttonEl.textContent = "🎙️"
            setStatus("Ready", "")
        }

        const setStatus = (text, cls) => {
            statusEl.textContent = text
            statusEl.className = cls || ""
        }

        buttonEl.addEventListener("click", () => {
            if (!connected) connect()
            else disconnect()
        })
    </script>
</body>
</html>

pyproject.toml

[project]
name = "felix-voice"
version = "0.1.0"
description = "Voice chat with Felix via Pipecat + Clawdbot"
requires-python = ">=3.10,<3.14"
dependencies = [
    "pipecat-ai[silero,deepgram,openai,elevenlabs,webrtc]",
    "python-dotenv",
    "loguru",
]

server.py

"""FastAPI signaling server for Felix Voice Chat."""

import argparse
import sys
from contextlib import asynccontextmanager

import uvicorn
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI
from fastapi.responses import FileResponse
from loguru import logger
from pipecat.transports.smallwebrtc.request_handler import (
    SmallWebRTCPatchRequest,
    SmallWebRTCRequest,
    SmallWebRTCRequestHandler,
)

from bot import run_bot

load_dotenv(override=True)


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await small_webrtc_handler.close()


app = FastAPI(lifespan=lifespan)
small_webrtc_handler = SmallWebRTCRequestHandler()


@app.post("/api/offer")
async def offer(request: SmallWebRTCRequest, background_tasks: BackgroundTasks):
    async def webrtc_connection_callback(connection):
        background_tasks.add_task(run_bot, connection)

    answer = await small_webrtc_handler.handle_web_request(
        request=request,
        webrtc_connection_callback=webrtc_connection_callback,
    )
    return answer


@app.patch("/api/offer")
async def ice_candidate(request: SmallWebRTCPatchRequest):
    await small_webrtc_handler.handle_patch_request(request)
    return {"status": "success"}


@app.get("/")
async def serve_index():
    return FileResponse("index.html")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Felix Voice Chat")
    parser.add_argument("--host", default="localhost", help="Host (default: localhost)")
    parser.add_argument("--port", type=int, default=7860, help="Port (default: 7860)")
    parser.add_argument("--verbose", "-v", action="count")
    args = parser.parse_args()

    logger.remove(0)
    if args.verbose:
        logger.add(sys.stderr, level="TRACE")
    else:
        logger.add(sys.stderr, level="DEBUG")

    print(f"\n📋 Felix Voice Chat")
    print(f"   Open http://localhost:{args.port} in your browser\n")

    uvicorn.run(app, host=args.host, port=args.port)

Porte 0 . Disco duro 255.0.0.0