Files
hold-slayer/config.py
Robert Helewka 94fb6cd79d feat: mount MCP server, add bearer auth, and guard outbound calls
The MCP server was created but never mounted — no client could reach
it. Mount it at /mcp/ over streamable HTTP with a combined lifespan,
resolving the gateway lazily so mounting happens at app construction.

Security and safety for the agent surface:
- One static API_TOKEN (SecretStr) enforced across REST (dependency),
  WebSocket (query param/header before accept), and MCP
  (StaticTokenVerifier). Startup refuses tokenless non-loopback binds.
- Emergency numbers (911/9911/112) always refused on make_call, plus a
  MAX_CONCURRENT_CALLS cap; ValueError surfaces as 400/ToolError.
- Safe defaults: debug off, no credential in default DATABASE_URL,
  SIP/LLM/TTS secrets as SecretStr.

Cleanups:
- Delete broken learn_call_flow tool (wrong ctor args, nonexistent
  method) and the never-fed CallAnalytics service; keep
  call_flow_learner for proper wiring later.
- Trim dial_plan to what is actually used (emergency guard, extension
  allocation); delete the unreferenced matcher/normaliser.
- Register call_history before calls so /api/calls/history is no
  longer shadowed by /api/calls/{call_id}.
- fastmcp pinned >=3.0 (http_app + StaticTokenVerifier).

New tests: MCP in-memory client (tool surface, lazy gateway, emergency
refusal, call cap) and API security (401 paths, route order, mount).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 15:20:24 -04:00

160 lines
4.7 KiB
Python

"""
Hold Slayer Gateway — Configuration
All settings loaded from environment variables / .env file.
"""
from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
class SIPTrunkSettings(BaseSettings):
"""SIP trunk provider configuration."""
model_config = SettingsConfigDict(env_prefix="SIP_TRUNK_")
host: str = "sip.provider.com"
port: int = 5060
username: str = ""
password: SecretStr = SecretStr("")
transport: str = "udp" # udp, tcp, tls
did: str = "" # Your phone number (E.164)
class GatewaySIPSettings(BaseSettings):
"""Gateway SIP listener for device registration."""
model_config = SettingsConfigDict(env_prefix="GATEWAY_SIP_")
host: str = "0.0.0.0"
port: int = 5060
domain: str = "gateway.local"
class SpeachesSettings(BaseSettings):
"""Speaches STT service configuration."""
model_config = SettingsConfigDict(env_prefix="SPEACHES_")
url: str = "http://localhost:22070"
model: str = "whisper-large-v3"
class ClassifierSettings(BaseSettings):
"""Audio classifier thresholds."""
model_config = SettingsConfigDict(env_prefix="CLASSIFIER_")
music_threshold: float = 0.7
speech_threshold: float = 0.6
silence_threshold: float = 0.85
window_seconds: float = 3.0
class LLMSettings(BaseSettings):
"""LLM service configuration (OpenAI-compatible API)."""
model_config = SettingsConfigDict(env_prefix="LLM_")
base_url: str = "http://localhost:11434/v1"
model: str = "llama3"
api_key: SecretStr = SecretStr("not-needed")
timeout: float = 30.0
max_tokens: int = 1024
temperature: float = 0.3
class HoldSlayerSettings(BaseSettings):
"""Hold Slayer behavior settings."""
model_config = SettingsConfigDict(env_prefix="HOLD_SLAYER_", env_prefix_allow_empty=True)
default_transfer_device: str = Field(
default="sip_phone", validation_alias="DEFAULT_TRANSFER_DEVICE"
)
max_hold_time: int = Field(default=7200, validation_alias="MAX_HOLD_TIME")
hold_check_interval: float = Field(default=2.0, validation_alias="HOLD_CHECK_INTERVAL")
class TTSSettings(BaseSettings):
"""Rhema TTS service configuration (OpenAI-compatible /v1/audio/speech)."""
model_config = SettingsConfigDict(env_prefix="TTS_")
base_url: str = "http://localhost:8000"
model: str = "speaches-ai/Kokoro-82M-v1.0-ONNX"
voice: str = "af_heart"
api_key: SecretStr = SecretStr("")
timeout: float = 30.0
sample_rate: int = 16000
class ReceptionistSettings(BaseSettings):
"""AI Receptionist behavior settings."""
model_config = SettingsConfigDict(env_prefix="RECEPTIONIST_")
enabled: bool = True
greeting_template: str = (
"Hi, you've reached Robert's line. Who's calling, and what's this about?"
)
message_prompt: str = "Please leave your message after the tone."
listen_timeout_s: float = 15.0
end_of_utterance_silence_s: float = 1.2
message_max_seconds: int = 90
llm_persona: str = (
"You are a helpful, concise phone receptionist. Decide whether to ring "
"the owner, take a message, or politely decline."
)
class Settings(BaseSettings):
"""Root application settings."""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore",
)
# Database — no default credentials; must be set in the environment
database_url: str = ""
# Server
host: str = "0.0.0.0"
port: int = 8000
debug: bool = False
log_level: str = "info"
# Auth — one static bearer token shared by REST, WebSocket, and MCP.
# Empty disables auth, which is only permitted on loopback binds.
api_token: SecretStr = SecretStr("")
# Outbound-call safety cap (REST + MCP make_call)
max_concurrent_calls: int = 4
# Notifications
notify_sms_number: str = ""
# Sub-configs
sip_trunk: SIPTrunkSettings = Field(default_factory=SIPTrunkSettings)
gateway_sip: GatewaySIPSettings = Field(default_factory=GatewaySIPSettings)
speaches: SpeachesSettings = Field(default_factory=SpeachesSettings)
classifier: ClassifierSettings = Field(default_factory=ClassifierSettings)
llm: LLMSettings = Field(default_factory=LLMSettings)
hold_slayer: HoldSlayerSettings = Field(default_factory=HoldSlayerSettings)
tts: TTSSettings = Field(default_factory=TTSSettings)
receptionist: ReceptionistSettings = Field(default_factory=ReceptionistSettings)
# Singleton
_settings: Settings | None = None
def get_settings() -> Settings:
"""Get cached application settings."""
global _settings
if _settings is None:
_settings = Settings()
return _settings