"""Main MCP server implementation for ElevenLabs integration."""
import json
import logging
import asyncio
from typing import Dict, Any, Optional, List
from mcp.server.fastmcp import FastMCP
from mcp.server.stdio import stdio_server
from .client import ElevenLabsClient, ElevenLabsError
from .config import settings
# Configure logging
logging.basicConfig(level=getattr(logging, settings.log_level.upper()))
logger = logging.getLogger(__name__)
# Create FastMCP server instance
mcp = FastMCP(
name=settings.server_name,
version=settings.server_version
)
# Global ElevenLabs client
elevenlabs_client: Optional[ElevenLabsClient] = None
async def get_client() -> ElevenLabsClient:
"""Get or create ElevenLabs client instance."""
global elevenlabs_client
if elevenlabs_client is None:
elevenlabs_client = ElevenLabsClient()
return elevenlabs_client
# Agent Management Tools
@mcp.tool()
async def create_agent(
conversation_config: Dict[str, Any],
platform_settings: Dict[str, Any] = None,
name: str = "New Agent"
) -> str:
"""Create a new ElevenLabs conversational AI agent.
Args:
conversation_config: Agent configuration including TTS, ASR, and prompt settings
platform_settings: Platform-specific settings (optional)
name: Human-readable name for the agent
Returns:
JSON string with agent creation result
"""
try:
client = await get_client()
config = {
"conversation_config": conversation_config,
"name": name
}
if platform_settings:
config["platform_settings"] = platform_settings
result = await client.create_agent(config)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Agent creation failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def get_agent(agent_id: str) -> str:
"""Get agent configuration by ID.
Args:
agent_id: Unique identifier for the agent
Returns:
JSON string with agent configuration
"""
try:
client = await get_client()
result = await client.get_agent(agent_id)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Get agent failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def list_agents(cursor: str = None, page_size: int = 30) -> str:
"""List all agents with pagination.
Args:
cursor: Pagination cursor (optional)
page_size: Number of agents to return (default: 30)
Returns:
JSON string with agents list
"""
try:
client = await get_client()
result = await client.list_agents(cursor, page_size)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"List agents failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def update_agent(
agent_id: str,
conversation_config: Dict[str, Any] = None,
platform_settings: Dict[str, Any] = None,
name: str = None,
tags: List[str] = None
) -> str:
"""Update existing agent configuration.
Args:
agent_id: Unique identifier for the agent
conversation_config: Updated agent configuration (optional)
platform_settings: Updated platform settings (optional)
name: Updated agent name (optional)
tags: Updated tags (optional)
Returns:
JSON string with update result
"""
try:
client = await get_client()
config = {}
if conversation_config:
config["conversation_config"] = conversation_config
if platform_settings:
config["platform_settings"] = platform_settings
if name:
config["name"] = name
if tags:
config["tags"] = tags
result = await client.update_agent(agent_id, config)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Update agent failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def delete_agent(agent_id: str) -> str:
"""Delete an agent.
Args:
agent_id: Unique identifier for the agent
Returns:
JSON string with deletion result
"""
try:
client = await get_client()
result = await client.delete_agent(agent_id)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Delete agent failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
# Tool Management Tools
@mcp.tool()
async def create_tool(
tool_type: str,
name: str,
description: str,
parameters: List[Dict[str, Any]] = None,
url: str = None,
method: str = "GET",
headers: Dict[str, str] = None,
wait_for_response: bool = True
) -> str:
"""Create a new tool for agents.
Args:
tool_type: Type of tool ('webhook' or 'client')
name: Tool name
description: Tool description
parameters: Tool parameters schema
url: Webhook URL (required for webhook tools)
method: HTTP method (for webhook tools)
headers: HTTP headers (for webhook tools)
wait_for_response: Whether to wait for response (for client tools)
Returns:
JSON string with tool creation result
"""
try:
client = await get_client()
tool_config = {
"name": name,
"description": description,
"type": tool_type
}
if tool_type == "webhook":
if not url:
return json.dumps({"error": "URL is required for webhook tools"}, indent=2)
tool_config["api_schema"] = {
"url": url,
"method": method
}
if headers:
tool_config["api_schema"]["request_headers"] = headers
if parameters:
tool_config["api_schema"]["request_body_schema"] = {
"type": "object",
"properties": {param["name"]: {"type": param["type"], "description": param["description"]} for param in parameters},
"required": [param["name"] for param in parameters if param.get("required", False)]
}
elif tool_type == "client":
tool_config["wait_for_response"] = wait_for_response
if parameters:
tool_config["parameters"] = parameters
result = await client.create_tool(tool_config)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Create tool failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def get_tool(tool_id: str) -> str:
"""Get tool configuration by ID.
Args:
tool_id: Unique identifier for the tool
Returns:
JSON string with tool configuration
"""
try:
client = await get_client()
result = await client.get_tool(tool_id)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Get tool failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def list_tools(tool_type: str = None, limit: int = None) -> str:
"""List all tools.
Args:
tool_type: Filter by tool type ('webhook' or 'client')
limit: Maximum number of tools to return
Returns:
JSON string with tools list
"""
try:
client = await get_client()
result = await client.list_tools(tool_type, limit)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"List tools failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def update_tool(
tool_id: str,
name: str = None,
description: str = None,
parameters: List[Dict[str, Any]] = None,
url: str = None,
method: str = None,
headers: Dict[str, str] = None
) -> str:
"""Update existing tool configuration.
Args:
tool_id: Unique identifier for the tool
name: Updated tool name (optional)
description: Updated tool description (optional)
parameters: Updated tool parameters (optional)
url: Updated webhook URL (optional)
method: Updated HTTP method (optional)
headers: Updated HTTP headers (optional)
Returns:
JSON string with update result
"""
try:
client = await get_client()
tool_config = {}
if name:
tool_config["name"] = name
if description:
tool_config["description"] = description
if parameters:
tool_config["parameters"] = parameters
if url:
tool_config["url"] = url
if method:
tool_config["method"] = method
if headers:
tool_config["headers"] = headers
result = await client.update_tool(tool_id, tool_config)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Update tool failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def delete_tool(tool_id: str) -> str:
"""Delete a tool.
Args:
tool_id: Unique identifier for the tool
Returns:
JSON string with deletion result
"""
try:
client = await get_client()
result = await client.delete_tool(tool_id)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Delete tool failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
# Knowledge Base Management Tools
@mcp.tool()
async def create_knowledge_base_from_text(
text: str,
name: str,
description: str = ""
) -> str:
"""Create knowledge base document from text.
Args:
text: Text content for the knowledge base
name: Name of the knowledge base document
description: Description of the document (optional)
Returns:
JSON string with creation result
"""
try:
client = await get_client()
result = await client.create_knowledge_base_from_text(text, name, description)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Create knowledge base from text failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def create_knowledge_base_from_url(
url: str,
name: str,
description: str = ""
) -> str:
"""Create knowledge base document from URL.
Args:
url: URL to scrape for knowledge base content
name: Name of the knowledge base document
description: Description of the document (optional)
Returns:
JSON string with creation result
"""
try:
client = await get_client()
result = await client.create_knowledge_base_from_url(url, name, description)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Create knowledge base from URL failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def get_knowledge_base_document(document_id: str) -> str:
"""Get knowledge base document by ID.
Args:
document_id: Unique identifier for the document
Returns:
JSON string with document details
"""
try:
client = await get_client()
result = await client.get_knowledge_base_document(document_id)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Get knowledge base document failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def list_knowledge_base_documents(
name_filter: str = None,
document_types: List[str] = None,
limit: int = None
) -> str:
"""List knowledge base documents.
Args:
name_filter: Filter documents by name prefix (optional)
document_types: Filter by document types (optional)
limit: Maximum number of documents to return (optional)
Returns:
JSON string with documents list
"""
try:
client = await get_client()
result = await client.list_knowledge_base_documents(name_filter, document_types, limit)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"List knowledge base documents failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def update_knowledge_base_document(
document_id: str,
name: str = None,
description: str = None
) -> str:
"""Update knowledge base document.
Args:
document_id: Unique identifier for the document
name: Updated document name (optional)
description: Updated document description (optional)
Returns:
JSON string with update result
"""
try:
client = await get_client()
result = await client.update_knowledge_base_document(document_id, name, description)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Update knowledge base document failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def delete_knowledge_base_document(document_id: str) -> str:
"""Delete knowledge base document.
Args:
document_id: Unique identifier for the document
Returns:
JSON string with deletion result
"""
try:
client = await get_client()
result = await client.delete_knowledge_base_document(document_id)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Delete knowledge base document failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def compute_rag_index(document_id: str) -> str:
"""Compute RAG index for a knowledge base document.
Args:
document_id: Unique identifier for the document
Returns:
JSON string with RAG index computation result
"""
try:
client = await get_client()
result = await client.compute_rag_index(document_id)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Compute RAG index failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
@mcp.tool()
async def get_document_content(document_id: str) -> str:
"""Get full content of a knowledge base document.
Args:
document_id: Unique identifier for the document
Returns:
JSON string with document content
"""
try:
client = await get_client()
result = await client.get_document_content(document_id)
return json.dumps(result, indent=2)
except ElevenLabsError as e:
logger.error(f"Get document content failed: {e}")
return json.dumps({"error": str(e)}, indent=2)
# Resource definitions
@mcp.resource("elevenlabs://agents")
def list_agents_resource() -> str:
"""List all ElevenLabs agents."""
return "Use the list_agents tool to get all agents"
@mcp.resource("elevenlabs://tools")
def list_tools_resource() -> str:
"""List all ElevenLabs tools."""
return "Use the list_tools tool to get all tools"
@mcp.resource("elevenlabs://knowledge-base")
def list_knowledge_base_resource() -> str:
"""List all ElevenLabs knowledge base documents."""
return "Use the list_knowledge_base_documents tool to get all knowledge base documents"
def main():
"""Main entry point for the MCP server."""
logger.info(f"Starting {settings.server_name} v{settings.server_version}")
try:
# Run the server using stdio transport
mcp.run(transport="stdio")
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Server error: {e}")
raise
finally:
logger.info("Server shutdown complete")
if __name__ == "__main__":
main()
