Voice Agent

# OpenAIRealtime — OpenAI Realtime API 기반
pip install clawops[agent,openai]

# GeminiRealtime — Google Gemini Live API 기반
pip install clawops[agent,gemini]

# PipelineSession — STT, LLM, TTS를 각각 선택하여 조합
pip install clawops[agent,deepgram,elevenlabs,openai]

# Pipeline + Anthropic Claude LLM
pip install clawops[agent,deepgram,elevenlabs,anthropic-llm]

# Pipeline + Google Gemini LLM
pip install clawops[agent,deepgram,elevenlabs,gemini]

# MCP 서버 연동
pip install clawops[agent,mcp]

# OpenTelemetry 트레이싱
pip install clawops[tracing]

# 조합 예시: Gemini + MCP + 트레이싱
pip install clawops[agent,gemini,mcp,tracing]

# 모든 extras 한번에 설치
pip install clawops[agent-all]

export CLAWOPS_API_KEY="your-api-key"
export CLAWOPS_ACCOUNT_ID="your-account-id"
export OPENAI_API_KEY="your-openai-key"         # OpenAI Realtime 또는 OpenAILLM
export ANTHROPIC_API_KEY="your-anthropic-key"   # AnthropicLLM
export GOOGLE_API_KEY="your-google-key"         # Gemini Realtime 또는 GeminiLLM (Google AI)
# 또는 Google Cloud Vertex AI 사용 시 (GOOGLE_API_KEY 불필요)
# export GOOGLE_GENAI_USE_VERTEXAI=true
# export GOOGLE_CLOUD_PROJECT="your-project-id"
# export GOOGLE_CLOUD_LOCATION="us-central1"
export MISTRAL_API_KEY="your-mistral-key"       # MistralLLM
export GROQ_API_KEY="your-groq-key"             # GroqLLM
export PERPLEXITY_API_KEY="your-perplexity-key" # PerplexityLLM
export TOGETHER_API_KEY="your-together-key"     # TogetherLLM
export FIREWORKS_API_KEY="your-fireworks-key"   # FireworksLLM
export DEEPSEEK_API_KEY="your-deepseek-key"     # DeepSeekLLM
export XAI_API_KEY="your-xai-key"               # XaiLLM
export DEEPGRAM_API_KEY="your-deepgram-key"     # Pipeline: DeepgramSTT
export ELEVENLABS_API_KEY="your-elevenlabs-key" # Pipeline: ElevenLabsTTS

from clawops.agent import ClawOpsAgent, OpenAIRealtime
import asyncio

agent = ClawOpsAgent(
    from_="07012341234",
    session=OpenAIRealtime(
        system_prompt="친절한 고객 상담원입니다.",
        voice="marin",
        model="gpt-realtime-1.5",
        language="ko",
        turn_detection={"type": "semantic_vad", "eagerness": "medium"},
        greeting=True,
    ),
)

asyncio.run(agent.serve())

from clawops.agent import ClawOpsAgent, GeminiRealtime
import asyncio

agent = ClawOpsAgent(
    from_="07012341234",
    session=GeminiRealtime(
        system_prompt="친절한 고객 상담원입니다.",
        voice="Kore",
        language="ko",
    ),
)

asyncio.run(agent.serve())

session=GeminiRealtime(
    system_prompt="상담원입니다.",
    voice="Kore",
    realtime_input_config={
        "automatic_activity_detection": {
            "start_of_speech_sensitivity": "START_SENSITIVITY_HIGH",
            "end_of_speech_sensitivity": "END_SENSITIVITY_HIGH",
            "prefix_padding_ms": 20,
            "silence_duration_ms": 100,
        },
        "activity_handling": "NO_INTERRUPTION",
    },
)

from clawops.agent import ClawOpsAgent
from clawops.agent.pipeline import PipelineSession, DeepgramSTT, OpenAILLM, ElevenLabsTTS
import asyncio

agent = ClawOpsAgent(
    from_="07012341234",
    session=PipelineSession(
        system_prompt="친절한 고객 상담원입니다.",
        stt=DeepgramSTT(language="ko"),
        llm=OpenAILLM(model="gpt-4o-mini"),
        tts=ElevenLabsTTS(voice_id="EXAVITQu4vr4xnSDxMaL"),
    ),
)

asyncio.run(agent.serve())

from collections.abc import AsyncIterator
from clawops.agent.pipeline import SpeechEvent

class MySTT:
    async def transcribe(self, audio_stream: AsyncIterator[bytes]) -> AsyncIterator[SpeechEvent]:
        # audio_stream: PCM16 signed 16-bit LE, 16kHz, mono
        # SpeechEvent(type="interim", transcript="...") → barge-in 트리거
        # SpeechEvent(type="final", transcript="...")   → 확정된 텍스트
        ...

class MyLLM:
    async def generate(
        self, messages: list[dict], tools: list[dict] | None = None,
    ) -> AsyncIterator[str]:
        # OpenAI Chat Completions 형식의 messages + tools
        # 텍스트 토큰을 스트리밍으로 yield
        # Tool 호출 시: yield '{"type":"tool_calls","tool_calls":[...]}'
        ...

class MyTTS:
    sample_rate: int = 24000  # 리샘플링을 위한 출력 샘플레이트

    async def synthesize(self, text_stream: AsyncIterator[str]) -> AsyncIterator[bytes]:
        # text_stream: 문장 단위로 분할된 텍스트
        # PCM16 오디오 청크를 yield (sample_rate에 맞는)
        ...

@agent.tool
async def check_order(order_id: str) -> str:
    """주문 상태를 확인합니다. 고객이 주문 번호를 말하면 이 함수를 호출하세요."""
    return f"주문 {order_id}은 배송 중입니다."

@agent.tool
async def get_store_hours(branch: str) -> str:
    """매장 영업시간을 조회합니다."""
    hours = {"강남점": "10:00-22:00", "홍대점": "11:00-23:00"}
    return hours.get(branch, "해당 매장을 찾을 수 없습니다.")

from clawops.agent import ClawOpsAgent, BuiltinTool, OpenAIRealtime

# 기본: 모든 내장 도구 활성화 (BuiltinTool.ALL)
agent = ClawOpsAgent(from_="07012341234", session=session)

# 내장 도구 전부 비활성화 (커스텀 도구만 사용)
agent = ClawOpsAgent(from_="07012341234", session=session, builtin_tools=BuiltinTool.NONE)

# hang_up만 사용 (DTMF 도구 제외)
agent = ClawOpsAgent(from_="07012341234", session=session, builtin_tools=[BuiltinTool.HANG_UP])

# DTMF만 사용, AI가 전화를 끊지 못하게
agent = ClawOpsAgent(
    from_="07012341234",
    session=session,
    builtin_tools=[BuiltinTool.COLLECT_DTMF, BuiltinTool.SEND_DTMF],
)

# AI가 자동으로 호출 (시스템 프롬프트에 전환 조건 명시)
# 또는 코드에서 직접 호출:
await call.transfer("01012345678")

await call.transfer(
    "01012345678",
    mode="warm",
    whisper="VIP 고객님이십니다. 주문 번호는 A1234입니다.",
)

await call.transfer(
    "01012345678",
    after_transfer="return",  # 기본값: "terminate"
)

await call.transfer(
    "01012345678",
    mode="warm",
    whisper="VIP 고객입니다.",
    hold_media="moh",        # 고객 대기 음악 ("moh" 또는 "silence")
    context={
        "customer_name": "홍길동",
        "order_id": "ORD-20260325-001",
        "priority": "high",
    },
)

@agent.on("call_start")
async def on_call_start(call):
    print(f"전화 수신: {call.from_number} -> {call.to_number}")
    print(f"통화 ID: {call.call_id}")
    print(f"방향: {call.direction}")  # "inbound" 또는 "outbound"

@agent.on("call_end")
async def on_call_end(call):
    print(f"통화 종료: {call.call_id} (통화 시간: {call.duration:.1f}초)")

@agent.on("call_failed")
async def on_call_failed(call, reason):
    print(f"통화 실패: {call.call_id} ({reason})")

@agent.on("transcript")
async def on_transcript(call, role, text):
    print(f"[{role}] {text}")  # role: "user" 또는 "assistant"

from clawops.agent import ClawOpsAgent, OpenAIRealtime
from clawops.agent.mcp import MCPServerStdio, MCPServerHTTP

agent = ClawOpsAgent(
    from_="07012341234",
    session=OpenAIRealtime(
        system_prompt="다양한 도구를 활용하는 AI 상담원입니다.",
    ),
    mcp_servers=[
        MCPServerStdio(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-google"],
            env={"GOOGLE_API_KEY": "..."},
        ),
        MCPServerHTTP(
            url="https://mcp.example.com/sse",
            headers={"Authorization": "Bearer your-token"},
        ),
    ],
)

async def main():
    session = await agent.call("01012345678", timeout=60)
    print(f"발신 중... (call_id: {session.call_id}, status: {session.status})")
    await session.wait()       # 통화 종료까지 대기
    await agent.disconnect()

asyncio.run(main())

async def main():
    await agent.connect()
    session = await agent.call("01012345678")
    # agent는 인바운드 수신도 계속 처리

agent = ClawOpsAgent(
    from_="07012341234",
    session=OpenAIRealtime(system_prompt="고객 상담원입니다."),
    recording=True,
    recording_path="./recordings",
)

pip install clawops[tracing]
pip install opentelemetry-sdk opentelemetry-exporter-otlp

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from clawops.agent.tracing import TracingConfig

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
trace.set_tracer_provider(provider)

agent = ClawOpsAgent(
    from_="07012341234",
    session=OpenAIRealtime(system_prompt="상담원입니다."),
    tracing=TracingConfig(
        service_name="my-call-center",
        tracer_provider=provider,
    ),
)

call (call.id, call.from, call.to, call.duration_ms)
├── mcp.connect (mcp.server.type, mcp.tools.count)
├── llm.session (gen_ai.system, gen_ai.request.model)
│   └── llm.generation
├── tool.call (tool.name, tool.source, tool.duration_ms)
│   └── mcp.call_tool (mcp.tool.name, mcp.tool.is_error)
└── tool.call

from clawops.agent import ClawOpsAgent
from clawops.agent.pipeline import PipelineSession, DeepgramSTT, OpenAILLM, ElevenLabsTTS
from clawops.agent.mcp import MCPServerStdio
import asyncio
import logging

logging.basicConfig(level=logging.INFO)

agent = ClawOpsAgent(
    from_="07012341234",
    session=PipelineSession(
        system_prompt="""당신은 온라인 쇼핑몰의 고객 상담원입니다.
주문 조회, 반품 접수, 매장 안내를 도와줍니다.""",
        stt=DeepgramSTT(language="ko"),
        llm=OpenAILLM(model="gpt-4o-mini", temperature=0.8),
        tts=ElevenLabsTTS(voice_id="EXAVITQu4vr4xnSDxMaL"),
    ),
    recording=True,
    mcp_servers=[
        MCPServerStdio(command="npx", args=["-y", "@modelcontextprotocol/server-google"]),
    ],
)

@agent.tool
async def check_order(order_id: str) -> str:
    """주문 번호로 주문 상태를 조회합니다."""
    orders = {
        "ORD-001": "배송 중 (내일 도착 예정)",
        "ORD-002": "배송 완료",
    }
    return orders.get(order_id, f"주문 {order_id}을 찾을 수 없습니다.")

@agent.tool
async def request_return(order_id: str, reason: str) -> str:
    """반품을 접수합니다."""
    return f"주문 {order_id}의 반품이 접수되었습니다. 사유: {reason}"

@agent.on("call_start")
async def on_start(call):
    print(f"[통화 시작] {call.from_number} -> {call.to_number}")

@agent.on("call_end")
async def on_end(call):
    print(f"[통화 종료] {call.call_id} ({call.duration:.1f}초)")

@agent.on("transcript")
async def on_transcript(call, role, text):
    print(f"[{role}] {text}")

asyncio.run(agent.serve())

# Ctrl+C로 안전하게 종료
asyncio.run(agent.serve())

async def main():
    session = await agent.call("01012345678")
    await session.wait()
    await agent.disconnect()

asyncio.run(main())

async def main():
    await agent.connect()
    session = await agent.call("01012345678")
    # agent는 인바운드 수신도 계속 처리
    # 종료 시:
    await agent.disconnect()

import logging
logging.getLogger("clawops.agent").setLevel(logging.DEBUG)