# Voice Agent (/docs/voice-agent)

## Voice Agent

한국AI통신 Voice Agent SDK는 Python으로 AI 음성 에이전트를 구축할 수 있는 고수준 프레임워크입니다.
세 가지 세션 타입을 지원합니다:

  이 문서는 **에이전트를 직접 호스팅하는** SDK 경로입니다. 코드 없이 대시보드에서 만든
  에이전트는 한국AI통신이 대신 실행하므로 `agent.serve()` 가 필요 없습니다 →
  [대시보드 에이전트](/docs/voice-agent/managed)

- **OpenAI Realtime** — OpenAI Realtime API를 사용하는 올인원 방식
- **Gemini Realtime** — Google Gemini Live API를 사용하는 올인원 방식
- **Pipeline** — STT → LLM → TTS를 각각 선택하여 조합하는 모듈형 방식

  각 제공자의 검증 상태와 지원 현황은 [제공자 호환성](/docs/voice-agent/providers) 문서를 반드시
  확인하세요. 현재 **OpenAI Realtime만 검증 완료** 상태이며, 나머지 제공자는 검증 전입니다.

  SDK는 모든 클래스와 메서드에 타입 힌트와 docstring을 제공합니다. IDE 자동완성으로 파라미터를
  확인할 수 있습니다.

## 아키텍처

Voice Agent는 두 개의 WebSocket 연결로 동작합니다.

- **Control WebSocket** — 한국AI통신 서버와의 시그널링 채널. 에이전트가 시작되면 하나의 Control WS가 열리고, 인바운드/아웃바운드 전화 이벤트를 수신합니다. 연결이 끊어지면 지수 백오프(1초~30초)로 자동 재연결합니다.
- **Media WebSocket** — 통화별 오디오 스트림 채널. 전화가 연결될 때마다 독립적인 Media WS가 생성되어 실시간 양방향 오디오를 전송합니다. 오디오 포맷은 G.711 μ-law, 8kHz, mono입니다.

  에이전트가 한국AI통신 서버에 직접 연결하므로 ngrok 같은 터널링 도구가 필요 없습니다. 로컬 환경에서
  바로 개발하고 테스트할 수 있습니다.

## 설치

Voice Agent를 사용하려면 기본 SDK에 세션 타입별 extras를 추가 설치합니다.

### 세션 타입별 설치

```bash
# 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]
```

### 추가 extras

필요에 따라 extras를 추가할 수 있습니다. extras는 자유롭게 조합 가능합니다.

```bash
# 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]
```

### 환경변수

```bash
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
```

## OpenAI Realtime

OpenAI Realtime API를 사용하는 가장 간단한 설정입니다.

```python
from clawops.agent import ClawOpsAgent, OpenAIRealtime
import asyncio

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

asyncio.run(agent.serve())
```


  `agent.serve()`는 SIGINT/SIGTERM까지 블로킹되며, 시그널 수신 시 자동으로 `disconnect()`를
  호출합니다. 논블로킹 연결만 필요한 경우 `agent.connect()`를 사용하세요.

### OpenAI Realtime 음성

| 음성     | 설명                                     |
| -------- | ---------------------------------------- |
| `marin`  | 기본 음성 (여성, 따뜻하고 자연스러운 톤) |
| `ash`    | 차분한 남성 톤                           |
| `ballad` | 부드럽고 감성적인 톤                     |
| `coral`  | 밝고 친근한 톤                           |
| `sage`   | 차분하고 전문적인 톤                     |
| `verse`  | 생동감 있는 톤                           |

### OpenAI Realtime 모델

| 모델               | 설명                                          |
| ------------------ | --------------------------------------------- |
| `gpt-realtime-2` | 빠른 응답 속도, 일반적인 상담에 적합 (기본값) |
| `gpt-4o-realtime`  | 높은 추론 능력, 복잡한 대화에 적합            |

## Gemini Realtime

Google Gemini Live API를 사용한 음성 대화입니다.

```python
from clawops.agent import ClawOpsAgent, GeminiRealtime
import asyncio

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

asyncio.run(agent.serve())
```

> **Note:** 기본 모델이 `gemini-3.1-flash-live-preview`로 업데이트되었습니다. 이전 `gemini-2.5-flash-native-audio-preview-12-2025` 모델은 더 이상 지원되지 않습니다.

### VAD (Voice Activity Detection) 설정

Gemini Live API의 음성 감지 세부 설정을 `realtime_input_config`로 전달할 수 있습니다.
구조는 [google-genai SDK의 `RealtimeInputConfig`](https://ai.google.dev/gemini-api/docs/live-api/capabilities#configure-automatic-vad)를 그대로 따릅니다.

```python
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",
    },
)
```

| 파라미터 | 타입 | 설명 |
| --- | --- | --- |
| `realtime_input_config` | `dict` | Gemini VAD 설정. `automatic_activity_detection`, `activity_handling`, `turn_coverage` 등을 포함. |

### 입력 음성 전사 설정

발신자 음성 전사에 언어 힌트와 고유 명사 목록을 `input_audio_transcription` 으로 넘길 수 있습니다(`clawops` 0.66.0 부터).
구조는 [google-genai SDK의 `AudioTranscriptionConfig`](https://docs.cloud.google.com/gemini-enterprise-agent-platform/models/live-api/best-practices#transcription-language-code)를 그대로 따릅니다.

```python
session=GeminiRealtime(
    system_prompt="상담원입니다.",
    model="gemini-3.8-live",
    input_audio_transcription={
        "language_codes": ["ko-KR"],
        "custom_vocabulary": ["클로옵스", "ClawOps"],
    },
)
```

| 파라미터 | 타입 | 설명 |
| --- | --- | --- |
| `input_audio_transcription` | `dict` | 입력 음성 전사 설정. 지정하지 않으면 빈 설정(`{}`)을 보냅니다. |

| 키 | 설명 |
| --- | --- |
| `language_codes` | BCP-47 언어 코드 목록. 여러 개를 넣을 수 있습니다. |
| `custom_vocabulary` | 전사가 우선 인식할 용어 목록. 최대 1,000개이며, 넘으면 세션을 열 때 거절됩니다. |

`gemini-3.8-live`·`gemini-3.8-live-extended-thinking`·`gemini-3.1-flash-live-preview` 모두 두 키를 받습니다(2026-10-20 확인).
google-genai 2.13.0 이상이 필요합니다. Node SDK 는 `inputAudioTranscription`(`languageCodes`·`customVocabulary`)입니다(`@teamlearners/clawops` 0.54.0 부터).

### 모델과 thinking 설정

`model` 에는 Gemini Live API 모델명을 그대로 넣습니다. 새 모델이 나와도 SDK 업데이트 없이 이름만 바꾸면 됩니다.

```python
session=GeminiRealtime(
    system_prompt="상담원입니다.",
    model="gemini-3.8-live",
)
```

`gemini-3.8-live-extended-thinking` 처럼 생각 수준을 요구하는 모델은 `thinking_config` 로 넘깁니다(`clawops` 0.58.0 부터).
구조는 [google-genai SDK의 `ThinkingConfig`](https://ai.google.dev/gemini-api/docs/thinking)를 그대로 따릅니다.

```python
session=GeminiRealtime(
    system_prompt="상담원입니다.",
    model="gemini-3.8-live-extended-thinking",
    thinking_config={"thinking_level": "LOW"},
)
```

| 파라미터 | 타입 | 설명 |
| --- | --- | --- |
| `thinking_config` | `dict` | Gemini thinking 설정. `thinking_level`, `thinking_budget` 등을 포함. 지정하지 않으면 보내지 않습니다. |

받는 값은 모델마다 다릅니다(2026-09-29 확인).

| 모델 | `thinking_config` |
| --- | --- |
| `gemini-3.8-live-extended-thinking` | `thinking_level` 필수 — `LOW`·`MEDIUM`·`HIGH` (`MINIMAL` 은 거절) |
| `gemini-3.8-live` | `thinking_level` 은 거절, `thinking_budget` 만 받음. 보통은 지정하지 않습니다 |
| `gemini-3.1-flash-live-preview` | `thinking_level`·`thinking_budget` 모두 받음 |

모델이 받지 않는 값을 넣으면 세션을 열 때 Gemini 가 거절 사유와 함께 실패합니다.
생각 수준을 높일수록 응답 전 대기가 길어지므로, 전화 상담에서는 `LOW` 부터 시작하는 것을 권장합니다.

### 함수 호출 behavior 와 proactive audio (3.8 Live)

Gemini 3.8 Live 는 함수 호출 기본값이 비동기(`NON_BLOCKING`)입니다. 비동기 모드에서는 도구 결과를
기다리는 동안 모델이 계속 말하므로, 전화에서는 결과가 나올 때까지 모델이 기다리는 동기(`BLOCKING`)가
맞습니다. SDK 는 모델별 기본값을 함수 선언에 자동으로 싣고(`clawops` 0.60.0 부터), `tool_behavior` 로
덮어쓸 수 있습니다.

```python
session=GeminiRealtime(
    system_prompt="상담원입니다.",
    model="gemini-3.8-live",
    tool_behavior="BLOCKING",   # 생략하면 3.8-live 는 BLOCKING 이 기본
)
```

| 파라미터 | 타입 | 설명 |
| --- | --- | --- |
| `tool_behavior` | `"BLOCKING" \| "NON_BLOCKING"` | 모든 함수 선언(커스텀·MCP·내장)에 실리는 `behavior`. 생략하면 모델별 기본값. |
| `proactive_audio` | `bool` | `proactivity.proactive_audio`. 지정하지 않으면 보내지 않습니다. |
| `api_version` | `str` | google-genai `http_options.api_version`. 기본 `v1beta`. |

모델별 동작(2026-10-10, 모델 페이지·실측 기준):

| 모델 | `behavior` | proactive audio |
| --- | --- | --- |
| `gemini-3.8-live` | 서버 기본값 NON_BLOCKING. SDK 기본값은 **BLOCKING** 명시 | 영구 ON. `False` 는 거절 |
| `gemini-3.8-live-extended-thinking` | **NON_BLOCKING 만** 받음(BLOCKING 은 하드 에러). SDK 기본값 NON_BLOCKING | 영구 ON. `False` 는 거절 |
| `gemini-3.1-flash-live-preview` | 동기만. SDK 는 보내지 않음 | 미지원 |

`gemini-3.8-live-extended-thinking` 은 비동기 도구 실행만 가능해, 도구 결과를 기다리는 동안 모델이
말할 수 있습니다. 도구를 쓰는 전화 에이전트에는 `gemini-3.8-live` 를 권장합니다.

타이밍 문제를 보고할 때는 `CLAWOPS_GEMINI_DEBUG=1` 로 띄워 `[gemini-trace]` 로그를 함께 보내 주세요.

### 도구별 비동기 실행 (`behavior`·`scheduling`)

도구마다 모델이 결과를 기다릴지(`BLOCKING`) 기다리지 않고 대화를 이어 갈지(`NON_BLOCKING`) 정할 수 있습니다
(`clawops` 0.61.0 부터). 결과가 다음 문장인 조회는 BLOCKING, 저장·기록처럼 말할 게 없는 작업은 NON_BLOCKING 이 맞습니다.

```python
@agent.tool(behavior="NON_BLOCKING", scheduling="SILENT")
async def save_note(text: str):
    """통화 중 메모를 저장합니다."""
    await crm.save(text)           # 모델은 기다리지 않는다. None 을 돌려주면 결과를 말하지 않는다(SILENT).

@agent.tool                        # 생략 = 세션 기본값(3.8-live 는 BLOCKING)
async def lookup_order(order_id: str) -> str:
    """주문을 조회합니다."""
    return await orders.get(order_id)
```

| 값 | 의미 |
| --- | --- |
| `behavior="BLOCKING"` | 결과를 받을 때까지 모델이 기다립니다. 그동안 `hold_audio` 가 재생됩니다. |
| `behavior="NON_BLOCKING"` | 도구가 도는 동안 모델이 계속 대화합니다. 결과는 끝나는 즉시 보내고, hold audio 는 틀지 않습니다. 서버가 호출을 취소하면(끼어들기 등) 진행 중인 도구를 취소합니다. |
| `scheduling="SILENT"` | 결과를 말하지 않고 기억만 합니다. 핸들러가 `None` 을 돌려주면 기본값입니다. |
| `scheduling="WHEN_IDLE"` | 지금 하던 말을 마친 뒤 전달합니다(서버 기본값). |
| `scheduling="INTERRUPT"` | 하던 말을 끊고 바로 전달합니다. |

세션 전체 기본값은 `GeminiRealtime(tool_behavior=..., tool_response_scheduling=...)` 이고 도구별 값이 우선합니다.
내장 도구(`hang_up`·`collect_dtmf`·`send_dtmf`·`transfer_call`)는 통화 제어라 항상 BLOCKING 입니다.
`gemini-3.8-live-extended-thinking` 은 NON_BLOCKING 만 받고 `scheduling` 을 거절하므로, 이 모델에서는 모든 도구가
NON_BLOCKING 이 되고 `scheduling` 은 보내지 않습니다. `scheduling` 은 OpenAI 등 다른 세션에서는 무시됩니다.

### 긴 통화와 세션 수명

Gemini Live 는 WebSocket 연결이 약 10분에서 끝날 수 있고(끝나기 전 `go_away`), 압축을 켜지 않은 오디오 세션은
15분에서 끝납니다. SDK 는 접속 때마다 `session_resumption` 핸들을 보내 두고, 연결이 끝나거나 끊기면 그 핸들로
새 연결을 열어 대화를 이어 갑니다(`clawops` 0.62.0 부터, 기본 3회 시도). 재접속 구간에는 몇 초의 공백이 생길 수
있습니다. 핸들 재접속이 끝내 거절되면 새 세션으로 통화를 이어 가고(대화 기억은 잃음), 그것마저 실패하면 통화에
`session_lost` 이벤트를 낸 뒤 통화를 끊습니다.

15분을 넘는 통화라면 컨텍스트 압축을 켜야 세션 자체가 끝나지 않습니다.

```python
session=GeminiRealtime(
    system_prompt="상담원입니다.",
    model="gemini-3.8-live",
    context_window_compression={"sliding_window": {}},
)

@agent.on("call_start")
async def on_start(call):
    async def on_lost(call, reason: str):
        log.error("Gemini 세션 유실: %s", reason)
    call.on("session_lost", on_lost)
```

| 파라미터 | 타입 | 설명 |
| --- | --- | --- |
| `context_window_compression` | `dict` | Gemini `ContextWindowCompressionConfig`. 긴 통화에 `{"sliding_window": {}}` 권장. 지정하지 않으면 보내지 않습니다. |
| `session_resumption` | `bool` | 재접속 사용 여부. 기본 `True`. |
| `reconnect_max_attempts` | `int` | 재접속 시도 상한. 기본 3. 다 실패하면 `session_lost` 뒤 통화를 끊습니다. |

## Pipeline 모드

STT, LLM, TTS를 각각 지정하여 `PipelineSession`으로 조합합니다.
원하는 Provider를 자유롭게 선택할 수 있습니다.

```python
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())
```

### 빌트인 Provider

각 카테고리별 빌트인 Provider의 상세 파라미터는 개별 페이지를 참고하세요.

| 카테고리                     | Provider        | 설명                                                              |
| ---------------------------- | --------------- | ----------------------------------------------------------------- |
| [STT](/docs/voice-agent/stt) | `DeepgramSTT`   | Deepgram Nova 기반 실시간 음성 인식. VAD 지원                     |
| [LLM](/docs/voice-agent/llm) | `OpenAILLM`     | OpenAI Chat Completions 기반 스트리밍 텍스트 생성. Tool 호출 지원 |
| [LLM](/docs/voice-agent/llm) | `AnthropicLLM`  | Anthropic Claude 기반 스트리밍 텍스트 생성. Tool 호출 지원        |
| [LLM](/docs/voice-agent/llm) | `GeminiLLM`     | Google Gemini 기반 스트리밍 텍스트 생성. Tool 호출 지원           |
| [LLM](/docs/voice-agent/llm) | `OllamaLLM`     | Ollama 로컬 LLM 기반 스트리밍 텍스트 생성. Tool 호출 지원         |
| [LLM](/docs/voice-agent/llm) | `MistralLLM`    | Mistral AI 기반 스트리밍 텍스트 생성. Tool 호출 지원              |
| [LLM](/docs/voice-agent/llm) | `GroqLLM`       | Groq 초저지연 스트리밍 텍스트 생성. Tool 호출 지원                |
| [LLM](/docs/voice-agent/llm) | `PerplexityLLM` | Perplexity 웹 검색 기반 스트리밍 텍스트 생성                      |
| [LLM](/docs/voice-agent/llm) | `TogetherLLM`   | Together AI 기반 스트리밍 텍스트 생성. Tool 호출 지원             |
| [LLM](/docs/voice-agent/llm) | `FireworksLLM`  | Fireworks AI 기반 스트리밍 텍스트 생성. Tool 호출 지원            |
| [LLM](/docs/voice-agent/llm) | `DeepSeekLLM`   | DeepSeek 기반 스트리밍 텍스트 생성. Tool 호출 지원                |
| [LLM](/docs/voice-agent/llm) | `XaiLLM`        | xAI Grok 기반 스트리밍 텍스트 생성. Tool 호출 지원                |
| [TTS](/docs/voice-agent/tts) | `ElevenLabsTTS` | ElevenLabs WebSocket 기반 저지연 TTS. 자동 오디오 포맷 변환       |

### 오디오 변환 파이프라인

Pipeline 모드에서는 오디오 포맷이 자동으로 변환됩니다:

1. 수신: G.711 μ-law 8kHz → PCM16 → 16kHz 리샘플링 (STT 입력)
2. 송신: TTS 출력 (24kHz PCM16) → 8kHz 리샘플링 → G.711 μ-law

  오디오 변환은 SDK 내부에서 자동으로 처리됩니다. 개발자가 코덱이나 샘플레이트를 직접 다룰 필요가
  없습니다.

### Barge-In (끼어들기)

Pipeline 모드는 자동 barge-in을 지원합니다. AI가 응답 중일 때 사용자가 말하면:

1. Deepgram VAD의 `SpeechStarted` 이벤트로 즉시 감지
2. 현재 재생 중인 AI 오디오를 `clear_audio()`로 즉시 중단
3. 진행 중인 LLM/TTS 응답 생성을 취소
4. 사용자 발화가 확정되면 새로운 응답 생성 시작

Debounce 메커니즘(0.5초)이 적용되어, 사용자가 연속으로 짧게 말할 때 중복 응답이 생성되지 않습니다.

### 커스텀 Provider

Python Protocol 인터페이스를 구현하여 직접 Provider를 만들 수 있습니다.

```python
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에 맞는)
        ...
```


  커스텀 Provider의 `sample_rate` 속성을 설정하면 SDK가 자동으로 8kHz로 리샘플링합니다.

## Tool 등록

`@agent.tool` 데코레이터로 에이전트가 호출할 수 있는 함수를 등록합니다.
함수의 시그니처와 독스트링에서 OpenAI function tool 스키마가 자동 생성됩니다.
Realtime 모드와 Pipeline 모드 모두에서 동일하게 동작합니다.

```python
@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, "해당 매장을 찾을 수 없습니다.")
```


  Tool 함수는 반드시 `async` 함수여야 하며, 반환값은 `str` 타입이어야 합니다. 함수의 독스트링이
  AI에게 전달되는 도구 설명이 되므로, 명확하고 자세하게 작성하세요.

### 내장 Tool (Built-in Tools)

Agent는 통화 제어를 위한 내장 도구를 기본 제공합니다. `BuiltinTool`로 어떤 내장 도구를 활성화할지 제어할 수 있습니다.

| 도구            | 상수                        | 설명                                                                                                                                |
| --------------- | --------------------------- | ----------------------------------------------------------------------------------------------------------------------------------- |
| `hang_up`       | `BuiltinTool.HANG_UP`       | 전화를 종료합니다. AI가 대화 완료를 판단하면 자동으로 호출합니다.                                                                   |
| `collect_dtmf`  | `BuiltinTool.COLLECT_DTMF`  | 사용자의 키패드(DTMF) 입력을 수집합니다. 본인 인증, 메뉴 선택 등에 사용합니다. [키패드 입력 받기](#키패드-입력-받기)를 참고하세요. |
| `send_dtmf`     | `BuiltinTool.SEND_DTMF`     | DTMF 신호를 전송합니다. ARS 메뉴 탐색, 내선번호 입력 등에 사용합니다.                                                               |
| `transfer_call` | `BuiltinTool.TRANSFER_CALL` | 통화를 다른 번호로 전환합니다. 고객센터 연결, 상담원 전환 등에 사용합니다. Blind(즉시 전환)와 Warm(안내 후 전환) 모드를 지원합니다. |

기본적으로 모든 내장 도구가 활성화됩니다. `BuiltinTool.ALL`(전부 활성화)과 `BuiltinTool.NONE`(전부 비활성화) 상수를 사용하거나, 개별 도구를 리스트로 지정할 수 있습니다.

```python
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],
)
```


  ARS 아웃바운드 봇처럼 발신 후 ARS를 탐색해야 하는 경우 `[BuiltinTool.HANG_UP,
  BuiltinTool.SEND_DTMF]`만 활성화하면 불필요한 도구 없이 깔끔하게 구성할 수 있습니다.

### Call Transfer (통화 전환)

AI 에이전트가 통화 중 다른 번호로 전환할 수 있습니다. AI가 자동으로 `transfer_call` 도구를 호출하거나, 코드에서 직접 `call.transfer()`를 호출할 수 있습니다.

**Blind Transfer** — 고객을 즉시 대상 번호로 전환합니다. 전환이 시작되면 AI 세션은 종료됩니다.

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

**Warm Transfer** — 대상이 전화를 받으면 whisper 메시지를 먼저 들려준 후 고객과 연결합니다. 고객은 whisper를 들을 수 없으며, 연결 대기 중에는 `hold_media` 음원(기본: 통화 연결음)이 재생됩니다.

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

**전환 후 AI 복귀** — 전환된 통화가 끝나면 AI가 다시 고객과 대화를 이어갑니다. 예를 들어 전문 상담원과 통화 후 AI가 후속 안내를 하는 시나리오에 적합합니다.

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

`after_transfer="return"` 이면 `call.transfer()` 는 전환이 끝날 때까지 기다렸다가 결과를 돌려줍니다. `status` 는 `completed`·`no-answer`·`busy`·`failed`·`canceled` 중 하나입니다. 대상이 받지 않거나 전환에 실패해도 고객과의 통화는 유지되며, 다시 전환할 수 있습니다. 에이전트 연결이 끊겨 있으면 통화가 종료됩니다. 내장 도구 `transfer_call` 은 결과를 기다리지 않으므로 AI에게 전환 결과가 전달되지 않습니다.

**Context 전달** — 전환 대상에게 고객 정보 등 구조화 데이터를 webhook으로 전달할 수 있습니다.

```python
await call.transfer(
    "01012345678",
    mode="warm",
    whisper="VIP 고객입니다.",
    hold_media="moh",        # 전환 중 고객에게 ("moh" | "ringback" | "silence")
    context={
        "customer_name": "홍길동",
        "order_id": "ORD-20260325-001",
        "priority": "high",
    },
)
```

| 파라미터         | 타입   | 기본값        | 설명                                                                                   |
| ---------------- | ------ | ------------- | -------------------------------------------------------------------------------------- |
| `to`             | `str`  | (필수)        | 전환할 전화번호                                                                        |
| `mode`           | `str`  | `"blind"`     | `"blind"`: 즉시 전환, `"warm"`: whisper 후 전환                                        |
| `after_transfer` | `str`  | `"terminate"` | `"terminate"`: AI 세션 종료, `"return"`: 전환 통화 종료 후 AI가 다시 대화를 이어감     |
| `hold_media`     | `str`  | `"ringback"`  | 전환 중 고객에게 재생할 대기 음원. `"moh"`: 대기 음악, `"ringback"`: 통화 연결음, `"silence"`: 무음 |
| `whisper`        | `str`  | `None`        | Warm 모드에서 대상이 전화를 받았을 때 전달할 안내 메시지 (TTS). 고객에게는 들리지 않음 |
| `context`        | `dict` | `None`        | 전환 대상에게 webhook으로 전달할 구조화 데이터 (예: 고객 정보, 주문 번호 등)           |
| `caller_id`      | `str`  | `None`        | 전환 발신 시 표시할 발신자 번호 오버라이드                                             |
| `timeout`        | `int`  | `30`          | 대상 응답 대기 시간 (초). 초과 시 전환 실패 처리                                       |

## 키패드 입력 받기

발신자가 누른 키는 별도 설정 없이 모델에 `[DTMF 입력: 534]` 형태의 글자로 전달됩니다.
음성과 같은 대화로 들어가므로 말과 키패드를 함께 받을 수 있습니다.
아래 동작은 Python SDK 0.70.0·Node SDK 0.57.0 기준입니다.

### 전달되는 시점

| 조건 | 전달 시점 |
| ---- | --------- |
| 종료 키 `#` | 누르는 즉시 |
| 자리 수 | 정한 자리 수가 모이는 즉시 |
| 그 밖 | 마지막 키를 누르고 2초 뒤 |

셋 중 먼저 오는 조건으로 전달됩니다.

```text
534#      →  [DTMF 입력: 534]     # 은 값에 포함되지 않습니다
534       →  [DTMF 입력: 534]     마지막 키 뒤 2초
#         →  [DTMF 입력: #]       숫자 없이 # 만 누른 경우
```

### 설정

| Python | Node | 기본값 | 설명 |
| ------ | ---- | ------ | ---- |
| `passive_dtmf_debounce_ms` | `passiveDtmfDebounceMs` | `2000` | 마지막 키 뒤 전달까지 기다리는 시간(ms) |
| `passive_dtmf_finish_key` | `passiveDtmfFinishKey` | `"#"` | 종료 키. `None`/`null` 이면 `#` 도 일반 키로 전달됩니다 |
| `passive_dtmf_max_digits` | `passiveDtmfMaxDigits` | 없음 | 이 자리 수가 모이면 바로 전달합니다 |

```python
agent = ClawOpsAgent(
    from_="07012341234",
    session=session,
    passive_dtmf_finish_key="#",
    passive_dtmf_max_digits=4,
)
```

```typescript
const agent = new ClawOpsAgent({
  from: '07012341234',
  session,
  passiveDtmfFinishKey: '#',
  passiveDtmfMaxDigits: 4,
});
```


  정한 자리 수보다 많이 누르면 남는 키는 다음 값으로 전달됩니다. `passive_dtmf_max_digits=4` 에서
  `900101` 을 누르면 `9001` 과 `01` 로 나뉩니다. 받는 자리 수가 일정하지 않으면 자리 수를 정하지 마세요.

시스템 프롬프트에는 키패드 입력이 어떤 형태로 들어오는지 적어 둡니다.

```text
키패드 입력은 '[DTMF 입력: 숫자]' 형태로 전달됩니다.
동의 여부를 물을 때는 "동의하시면 1번, 동의하지 않으시면 2번을 눌러주세요"라고 안내하세요.
정류장 번호를 물을 때는 "정류장 번호를 누르시고 마지막에 우물정자를 눌러주세요"라고 안내하세요.
```

### `collect_dtmf` 도구

모델이 내장 도구 `collect_dtmf` 를 호출해 키를 직접 받을 수도 있습니다.

- 도구가 호출된 뒤 누른 키만 받습니다. 그 전에 누른 키는 위 규칙대로 모델에 전달됩니다.
- 기다리는 시간(`timeout`, 기본 5초)은 안내 음성 재생이 끝난 뒤부터 셉니다.
- 안내 음성이 나오는 중에 키를 누르면 안내를 멈추고 그 키부터 받습니다.
- 도구가 실행되는 동안에는 대기음(`tool_config.hold_audio`)을 틀지 않습니다.

모델이 이 도구를 쓰지 않게 하려면 `builtin_tools` 에서 `BuiltinTool.COLLECT_DTMF` 를 뺍니다.

### 키를 직접 처리하기

`dtmf` 이벤트는 키를 누를 때마다 하나씩, 누른 값 그대로 옵니다. 모델 전달과 별개로 항상 발생합니다.

```python
@agent.on("dtmf")
async def on_dtmf(call, digit):
    print(f"{call.call_id}: {digit}")
```

```typescript
agent.on('dtmf', (call, digit) => {
  console.log(`${call.callId}: ${digit}`);
});
```

## 이벤트 핸들링

`@agent.on()` 데코레이터로 통화 이벤트를 수신합니다.

```python
@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"
    if call.caller_id_restricted:
        print("발신자가 번호 표시제한을 걸고 건 통화")  # 번호는 from_number 에 그대로 옵니다
    if call.forwarded_from:
        print(f"착신전환 통화 — 원래 건 번호: {call.forwarded_from}")

@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"
```

`call.caller_id_restricted` 는 발신자가 번호 표시제한(`*23#` 등)을 걸고 건 수신 통화면 `True` 입니다.
번호는 `from_number` 에 그대로 전달되고, 표시제한 여부만 이 값으로 알 수 있습니다. 통신사가 표시제한 여부를
보내지 않은 통화와 발신 통화는 `False` 입니다. Python SDK 0.59.0·Node SDK 0.47.0(`call.callerIdRestricted`)부터 쓸 수 있습니다.

`call.forwarded_from` 은 통신사 착신전환으로 들어온 수신 통화에서 발신자가 **원래 건 번호**입니다. 휴대폰 등 다른
번호를 한국AI통신 번호로 착신전환해 두었을 때 어느 번호로 걸려온 전화인지 구분할 수 있습니다. 착신전환을 실행한
통신사가 전환 정보를 함께 보내온 경우에만 값이 있고, 직접 걸려온 통화와 발신 통화는 `None` 입니다. Python SDK 0.65.0·Node SDK
0.53.0(`call.forwardedFrom`, 없으면 `null`)부터 쓸 수 있습니다.

### 지원 이벤트

| 이벤트        | 설명                      | 핸들러 인자          |
| ------------- | ------------------------- | -------------------- |
| `call_start`  | 통화가 시작될 때          | `(call)`             |
| `call_end`    | 통화가 종료될 때          | `(call)`             |
| `call_failed` | 아웃바운드 통화 실패 시   | `(call, reason)`     |
| `transcript`  | 음성이 텍스트로 변환될 때 | `(call, role, text)` |
| `dtmf`        | 발신자가 키를 누를 때     | `(call, digit)`      |

### CallSession 메서드

| 메서드                               | 설명                                    |
| ------------------------------------ | --------------------------------------- |
| `await call.send_audio(ulaw_bytes)`  | G.711 μ-law(8kHz) 오디오를 전송. **PCM16 아님** |
| `await call.clear_audio()`           | 오디오 재생 큐를 즉시 비움 (barge-in)   |
| `await call.hangup()`                | 통화 종료                               |
| `await call.transfer(to, **opts)`    | 통화를 다른 번호로 전환 (blind/warm)    |
| `await call.wait()`                  | 통화 종료까지 대기 (아웃바운드 시 유용) |

### 받기 전에 거절하기 (`on_incoming`)

수신 통화를 **받기 전에** 보고 거절할 수 있습니다. 거절한 통화는 응답 전에 끝나므로 **과금되지 않고**,
세션·미디어가 열리지 않으며 `call_start`/`call_end` 등 이벤트도 발화되지 않습니다. 다른 에이전트나 웹훅으로 넘어가지도 않습니다.
Python SDK 0.63.0·Node SDK 0.51.0부터 쓸 수 있습니다.

```python
from clawops.agent import IncomingCall

@agent.on_incoming
async def screen(call: IncomingCall):
    if call.caller_id_restricted:          # 번호 표시제한 통화
        await call.reject()                # 거절
    elif call.from_number in blocklist:
        await call.reject(reason="busy")   # 통화중으로 거절
    # reject() 를 부르지 않고 끝나면 평소대로 받습니다.
```

```typescript
agent.onIncoming(async (call) => {
  if (call.callerIdRestricted) await call.reject({ reason: 'busy' });
});
```

`IncomingCall` 은 `call_id`·`from_number`·`to_number`·`caller_id_restricted`·`forwarded_from` 을 가집니다. 받기 전이라 오디오를 보내거나 전환할 수 없습니다.

| `reason` | 발신자에게 | 통화 기록 `status` · `hangupCause` | `sipResponseCode` |
| -------- | ---------- | ---------------------------------- | ----------------- |
| `"rejected"` (기본) | 통신사 안내 멘트 | `rejected` · `call_rejected` | 403 |
| `"busy"` | 통신사 통화중 안내 | `busy` · `user_busy` | 486 |

| 훅의 결과 | 처리 |
| --------- | ---- |
| `reject()` 호출 | 그 즉시 거절합니다 |
| `reject()` 없이 끝남 | 받습니다 (훅이 없을 때와 같음) |
| 예외 발생 | 경고 로그를 남기고 **받습니다** |
| 5초 안에 결정 없음 | 경고 로그를 남기고 **받습니다**. 그 뒤의 `reject()` 는 무시됩니다 |

특정 번호나 번호 표시제한 통화를 코드 없이 막으려면 [수신 통화 차단](/docs/call-blocking) 을 쓰세요.
차단 목록과 번호별 설정은 에이전트에 통화가 오기 전에, 벨을 울리지 않고 거절합니다.

## MCP 서버 연동

  MCP 연동을 사용하려면 `mcp` extras가 필요합니다: `pip install clawops[agent,mcp]`

[Model Context Protocol (MCP)](https://modelcontextprotocol.io) 서버를 연결하여 외부 도구를 AI 에이전트에 추가할 수 있습니다.
Stdio 방식과 HTTP 방식을 모두 지원합니다.

```python
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"},
        ),
    ],
)
```


  MCP 서버는 통화별로 독립적으로 연결됩니다. 동시에 여러 통화가 진행되어도 MCP tool 이름 충돌 없이
  안전하게 동작합니다.

## 아웃바운드 콜

에이전트에서 직접 발신 전화를 걸 수 있습니다. `agent.call()`은 `CallSession`을 즉시 반환하며, 상대방이 전화를 받으면 AI 대화가 자동으로 시작됩니다.

```python
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())
```

인바운드와 아웃바운드를 동시에 처리하려면 `connect()`로 연결 후 발신합니다:

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

## 오디오 녹음

`recording=True`로 설정하면 통화별로 오디오가 자동으로 파일에 저장됩니다.

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

녹음 파일은 통화별로 3개가 생성됩니다 (PCM16, 8kHz, mono WAV).

| 파일명              | 설명                        |
| ------------------- | --------------------------- |
| `{call_id}_in.wav`  | 상대방(발신자) 음성         |
| `{call_id}_out.wav` | AI 에이전트 음성            |
| `{call_id}_mix.wav` | 양쪽 음성이 혼합된 스테레오 |

  녹음 파일은 로컬 파일 시스템에 저장됩니다. 프로덕션 환경에서는 `recording_path`를 적절한 스토리지
  경로로 설정하고, 필요에 따라 S3 등 외부 스토리지로 업로드하는 로직을 추가하세요.

## 트레이싱 (OpenTelemetry)

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

```python
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,
    ),
)
```

### Span 계층 구조

```
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
```

## 전체 예제 (Pipeline 모드)

```python
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())
```

## 에이전트 종료

### 인바운드 서버 모드

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

### 아웃바운드 단건 발신

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

asyncio.run(main())
```

### 혼합 모드 (인바운드 + 아웃바운드)

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

## 디버그 로깅

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

## 에러 처리

| 예외                   | 설명                |
| ---------------------- | ------------------- |
| `AgentError`           | 기본 에이전트 예외  |
| `AgentConnectionError` | WebSocket 연결 실패 |