"""ElevenLabs API client for MCP server."""
import json
import logging
from typing import Dict, Any, Optional, List
from urllib.parse import urljoin
import httpx
from httpx import AsyncClient
from .config import settings
logger = logging.getLogger(__name__)
class ElevenLabsError(Exception):
"""Base exception for ElevenLabs API errors."""
pass
class ElevenLabsClient:
"""Async HTTP client for ElevenLabs Conversational AI API."""
def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
self.api_key = api_key or settings.elevenlabs_api_key
self.base_url = base_url or settings.elevenlabs_base_url
self.client = AsyncClient(
timeout=settings.request_timeout,
headers={
"xi-api-key": self.api_key,
"Content-Type": "application/json"
}
)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.client.aclose()
async def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Make HTTP request to ElevenLabs API."""
url = self.base_url.rstrip("/") + "/" + endpoint.lstrip("/")
try:
response = await self.client.request(method, url, **kwargs)
response.raise_for_status()
# Handle empty responses (like 204 No Content)
if response.status_code == 204 or not response.content:
return {"success": True}
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"HTTP error {e.response.status_code}: {e.response.text}")
raise ElevenLabsError(f"API request failed: {e.response.status_code}")
except httpx.RequestError as e:
logger.error(f"Request error: {e}")
raise ElevenLabsError(f"Network error: {e}")
except json.JSONDecodeError as e:
logger.error(f"JSON decode error: {e}")
raise ElevenLabsError(f"Invalid JSON response: {e}")
# Agent Management Methods
async def create_agent(self, config: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new conversational AI agent."""
return await self._request("POST", "/convai/agents/create", json=config)
async def get_agent(self, agent_id: str) -> Dict[str, Any]:
"""Get agent configuration by ID."""
return await self._request("GET", f"/convai/agents/{agent_id}")
async def list_agents(self, cursor: Optional[str] = None, page_size: int = 30) -> Dict[str, Any]:
"""List all agents with pagination."""
params = {"page_size": page_size}
if cursor:
params["cursor"] = cursor
return await self._request("GET", "/convai/agents", params=params)
async def update_agent(self, agent_id: str, config: Dict[str, Any]) -> Dict[str, Any]:
"""Update existing agent configuration."""
return await self._request("PATCH", f"/convai/agents/{agent_id}", json=config)
async def delete_agent(self, agent_id: str) -> Dict[str, Any]:
"""Delete an agent."""
return await self._request("DELETE", f"/convai/agents/{agent_id}")
# Tools Management Methods
async def create_tool(self, tool_config: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new tool."""
return await self._request("POST", "/convai/tools", json={"tool_config": tool_config})
async def get_tool(self, tool_id: str) -> Dict[str, Any]:
"""Get tool configuration by ID."""
return await self._request("GET", f"/convai/tools/{tool_id}")
async def list_tools(self, tool_type: Optional[str] = None, limit: Optional[int] = None) -> Dict[str, Any]:
"""List all tools."""
params = {}
if tool_type:
params["type"] = tool_type
if limit:
params["limit"] = limit
return await self._request("GET", "/convai/tools", params=params)
async def update_tool(self, tool_id: str, tool_config: Dict[str, Any]) -> Dict[str, Any]:
"""Update existing tool configuration."""
return await self._request("PATCH", f"/convai/tools/{tool_id}", json={"tool_config": tool_config})
async def delete_tool(self, tool_id: str) -> Dict[str, Any]:
"""Delete a tool."""
return await self._request("DELETE", f"/convai/tools/{tool_id}")
# Knowledge Base Management Methods
async def create_knowledge_base_from_text(self, text: str, name: str, description: str = "") -> Dict[str, Any]:
"""Create knowledge base document from text."""
return await self._request("POST", "/convai/knowledge-base", json={
"text": text,
"name": name,
"description": description
})
async def create_knowledge_base_from_url(self, url: str, name: str, description: str = "") -> Dict[str, Any]:
"""Create knowledge base document from URL."""
return await self._request("POST", "/convai/knowledge-base/url", json={
"url": url,
"name": name,
"description": description
})
async def get_knowledge_base_document(self, document_id: str) -> Dict[str, Any]:
"""Get knowledge base document by ID."""
return await self._request("GET", f"/convai/knowledge-base/{document_id}")
async def list_knowledge_base_documents(self, name_filter: Optional[str] = None,
document_types: Optional[List[str]] = None,
limit: Optional[int] = None) -> Dict[str, Any]:
"""List knowledge base documents."""
params = {}
if name_filter:
params["name_filter"] = name_filter
if document_types:
params["document_types"] = ",".join(document_types)
if limit:
params["limit"] = limit
return await self._request("GET", "/convai/knowledge-base", params=params)
async def update_knowledge_base_document(self, document_id: str, name: Optional[str] = None,
description: Optional[str] = None) -> Dict[str, Any]:
"""Update knowledge base document."""
data = {}
if name:
data["name"] = name
if description:
data["description"] = description
return await self._request("PATCH", f"/convai/knowledge-base/{document_id}", json=data)
async def delete_knowledge_base_document(self, document_id: str) -> Dict[str, Any]:
"""Delete knowledge base document."""
return await self._request("DELETE", f"/convai/knowledge-base/{document_id}")
async def compute_rag_index(self, document_id: str) -> Dict[str, Any]:
"""Compute RAG index for a document."""
return await self._request("POST", f"/convai/knowledge-base/{document_id}/compute-rag-index")
async def get_document_content(self, document_id: str) -> Dict[str, Any]:
"""Get full content of a knowledge base document."""
return await self._request("GET", f"/convai/knowledge-base/{document_id}/content")
async def upload_file_to_knowledge_base(self, file_data: bytes, filename: str,
name: str, description: str = "") -> Dict[str, Any]:
"""Upload file to create knowledge base document."""
files = {"file": (filename, file_data)}
data = {"name": name, "description": description}
# Remove Content-Type header for multipart upload
headers = {"xi-api-key": self.api_key}
try:
response = await self.client.post(
urljoin(self.base_url, "/convai/knowledge-base"),
headers=headers,
files=files,
data=data
)
response.raise_for_status()
return response.json()
except httpx.HTTPStatusError as e:
logger.error(f"File upload error {e.response.status_code}: {e.response.text}")
raise ElevenLabsError(f"File upload failed: {e.response.status_code}")
except httpx.RequestError as e:
logger.error(f"File upload request error: {e}")
raise ElevenLabsError(f"Network error: {e}")
async def close(self):
"""Close the HTTP client."""
await self.client.aclose()
