# Directory Structure
```
├── .env.example
├── .gitignore
├── .python-version
├── build_and_publish.sh
├── heygen_logo.png
├── heygen_mcp
│ ├── __init__.py
│ ├── api_client.py
│ └── server.py
├── pyproject.toml
├── README.md
├── run.py
├── setup.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
HEYGEN_API_KEY="<API_KEY_HERE>"
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# Environment
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# IDE
.idea/
.vscode/
*.swp
*.swo
# Logs
*.log
# Local development
.DS_Store
.ruff_cache/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Heygen MCP Server
The HeyGen MCP server enables any MCP Client like Claude Desktop or Agents to use the [HeyGen API](https://docs.heygen.com/) to generate avatars and videos.
[](https://opensource.org/licenses/MIT)
Note: This project is in early development. While we welcome community feedback and contributions, please be aware that official support is limited.
## Installation
### Prerequisites
- Python 3.10 or higher
- A Heygen API key (get one from [Heygen](https://www.heygen.com/)). Includes 10 Free Credits per Month
### Installing uv
uv is a fast Python package installer and resolver that we recommend for installing this package.
**macOS or Linux:**
```bash
# Install with the official installer script
curl -LsSf https://astral.sh/uv/install.sh | sh
# Or via Homebrew (macOS)
brew install uv
```
**Windows:**
```powershell
# Install with the official installer script in PowerShell
irm https://astral.sh/uv/install.ps1 | iex
# Or via Scoop
scoop install uv
```
For other installation methods, see the [uv documentation](https://github.com/astral-sh/uv).
## Usage
### Quickstart with Claude Desktop
1. Get your API key from [HeyGen](https://www.heygen.com/).
2. Install uv package manager (see [Installing uv](#installing-uv) section above).
3. Go to Claude > Settings > Developer > Edit Config > `claude_desktop_config.json` to include the following:
```json
{
"mcpServers": {
"HeyGen": {
"command": "uvx",
"args": ["heygen-mcp"],
"env": {
"HEYGEN_API_KEY": "<insert-your-api-key-here>"
}
}
}
}
```
If you're using Windows, you'll need to enable "Developer Mode" in Claude Desktop to use the MCP server. Click "Help" in the hamburger menu at the top left and select "Enable Developer Mode".
### Available MCP Tools
The server provides the following tools to Claude:
- **get_remaining_credits**: Retrieves the remaining credits in your Heygen account.
- **get_voices**: Retrieves a list of available voices from the Heygen API (limited to first 100 voices).
- **get_avatar_groups**: Retrieves a list of Heygen avatar groups.
- **get_avatars_in_avatar_group**: Retrieves a list of avatars in a specific Heygen avatar group.
- **generate_avatar_video**: Generates a new avatar video with the specified avatar, text, and voice.
- **get_avatar_video_status**: Retrieves the status of a video generated via the Heygen API.
## Development
### Running with MCP Inspector
To run the server locally with the MCP Inspector for testing and debugging:
```bash
uv --with "mcp[cli]" dev heygen_mcp/server.py
```
This will start the server in development mode and allow you to use the MCP Inspector to test the available tools and functionality.
## Roadmap
- [ ] Tests
- [ ] CICD
- [ ] Photo Avatar APIs Support
- [ ] SSE And Remote MCP Server with OAuth Flow
- [ ] Translation API Support
- [ ] Template API Support
- [ ] Interactive Avatar API Support
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
## License
This project is licensed under the MIT License - see the LICENSE file for details.
```
--------------------------------------------------------------------------------
/build_and_publish.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# Remove dist folder if it exists
rm -rf dist
rm -rf *.egg-info
# Build package
uv build
# Publish to PyPI
uv publish
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""Setup script for backwards compatibility with older pip versions."""
from setuptools import setup
if __name__ == "__main__":
setup()
```
--------------------------------------------------------------------------------
/run.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python
"""Simple entry point to run the heygen-mcp server during development."""
from heygen_mcp.server import main
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/heygen_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""HeyGen MCP - API client and MCP server for HeyGen API interaction."""
__version__ = "0.0.3"
from heygen_mcp.api_client import HeyGenApiClient
from heygen_mcp.server import main, mcp
__all__ = ["HeyGenApiClient", "mcp", "main"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["setuptools>=61.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "heygen-mcp"
version = "0.0.3"
description = "HeyGen MCP Server for AI Video Creation"
readme = "README.md"
requires-python = ">=3.12"
license = {text = "MIT"}
authors = [
{ name = "Eddy Kim", email = "[email protected]" },
]
keywords = ["heygen", "mcp", "claude", "ai", "video", "avatar"]
classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.12",
]
dependencies = [
"mcp[cli]>=1.6.0",
"pydantic>=2.0.0",
"httpx>=0.27.0",
"python-dotenv>=1.0.0",
]
[project.scripts]
heygen-mcp = "heygen_mcp.server:main"
[project.urls]
"Homepage" = "https://github.com/heygen-com/heygen-mcp"
"Bug Tracker" = "https://github.com/heygen-com/heygen-mcp/issues"
[project.optional-dependencies]
dev = [
"pytest",
"pytest-asyncio",
"ruff",
"build",
"twine",
]
[tool.ruff]
line-length = 88
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "B", "I"]
[tool.ruff.format]
quote-style = "double"
indent-style = "space"
[tool.uv]
# The commands subsection is not valid in tool.uv
# The entry point is already defined in [project.scripts]
```
--------------------------------------------------------------------------------
/heygen_mcp/server.py:
--------------------------------------------------------------------------------
```python
"""HeyGen MCP server module for providing MCP tools for the HeyGen API."""
import argparse
import os
import sys
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
from heygen_mcp.api_client import (
Character,
Dimension,
HeyGenApiClient,
MCPAvatarGroupResponse,
MCPAvatarsInGroupResponse,
MCPGetCreditsResponse,
MCPVideoGenerateResponse,
MCPVideoStatusResponse,
MCPVoicesResponse,
VideoGenerateRequest,
VideoInput,
Voice,
)
# Load environment variables
load_dotenv()
# Create MCP server instance
mcp = FastMCP("HeyGen MCP")
api_client = None
# Function to get or create API client
async def get_api_client() -> HeyGenApiClient:
"""Get the API client, creating it if necessary."""
global api_client
# If we already have a client, return it
if api_client is not None:
return api_client
# Otherwise, get the API key and create a new client
api_key = os.getenv("HEYGEN_API_KEY")
if not api_key:
raise ValueError("HEYGEN_API_KEY environment variable not set.")
# Create and store the client
api_client = HeyGenApiClient(api_key)
return api_client
########################
# MCP Tool Definitions #
########################
@mcp.tool(
name="get_remaining_credits",
description="Retrieves the remaining credits in heygen account.",
)
async def get_remaining_credits() -> MCPGetCreditsResponse:
"""Get the remaining quota for the user via HeyGen API."""
try:
client = await get_api_client()
return await client.get_remaining_credits()
except Exception as e:
return MCPGetCreditsResponse(error=str(e))
@mcp.tool(
name="get_voices",
description=(
"Retrieves a list of available voices from the HeyGen API. Results truncated "
"to first 100 voices. Private voices generally will returned 1st."
),
)
async def get_voices() -> MCPVoicesResponse:
"""Get the list of available voices via HeyGen API."""
try:
client = await get_api_client()
return await client.get_voices()
except Exception as e:
return MCPVoicesResponse(error=str(e))
@mcp.tool(
name="get_avatar_groups",
description=(
"Retrieves a list of HeyGen avatar groups. By default, only private avatar "
"groups are returned, unless include_public is set to true. Avatar groups "
"are collections of avatars, avatar group ids cannot be used to generate "
"videos."
),
)
async def get_avatar_groups(include_public: bool = False) -> MCPAvatarGroupResponse:
"""List avatar groups via HeyGen API v2/avatar_group.list endpoint."""
try:
client = await get_api_client()
return await client.list_avatar_groups(include_public)
except Exception as e:
return MCPAvatarGroupResponse(error=str(e))
@mcp.tool(
name="get_avatars_in_avatar_group",
description="Retrieves a list of avatars in a specific HeyGen avatar group.",
)
async def get_avatars_in_avatar_group(group_id: str) -> MCPAvatarsInGroupResponse:
"""List avatars in a specific HeyGen avatar group via HeyGen API."""
try:
client = await get_api_client()
return await client.get_avatars_in_group(group_id)
except Exception as e:
return MCPAvatarsInGroupResponse(error=str(e))
@mcp.tool(
name="generate_avatar_video",
description="Generates a new avatar video via the HeyGen API.",
)
async def generate_avatar_video(
avatar_id: str, input_text: str, voice_id: str, title: str = ""
) -> MCPVideoGenerateResponse:
"""Generate a new avatar video using the HeyGen API."""
try:
# Create the request object with default values
request = VideoGenerateRequest(
title=title,
video_inputs=[
VideoInput(
character=Character(avatar_id=avatar_id),
voice=Voice(input_text=input_text, voice_id=voice_id),
)
],
dimension=Dimension(width=1280, height=720),
)
client = await get_api_client()
return await client.generate_avatar_video(request)
except Exception as e:
return MCPVideoGenerateResponse(error=str(e))
@mcp.tool(
name="get_avatar_video_status",
description=(
"Retrieves the status of a video generated via the HeyGen API. Video status "
"make take several minutes to hours depending on length of video and queue "
"time. If video is not yet complete, status be viewed later by user via "
"https://app.heygen.com/home"
),
)
async def get_avatar_video_status(video_id: str) -> MCPVideoStatusResponse:
"""Retrieve the status of a video generated via the HeyGen API."""
try:
client = await get_api_client()
return await client.get_video_status(video_id)
except Exception as e:
return MCPVideoStatusResponse(error=str(e))
def parse_args():
"""Parse command-line arguments."""
parser = argparse.ArgumentParser(description="HeyGen MCP Server")
parser.add_argument(
"--api-key",
help=(
"HeyGen API key. Alternatively, set HEYGEN_API_KEY environment variable."
),
)
parser.add_argument(
"--host", default="127.0.0.1", help="Host to bind the server to."
)
parser.add_argument(
"--port", type=int, default=8000, help="Port to bind the server to."
)
parser.add_argument(
"--reload",
action="store_true",
help="Enable auto-reload for development.",
)
return parser.parse_args()
def main():
"""Run the MCP server."""
args = parse_args()
# Check if API key is provided or in environment
if args.api_key:
os.environ["HEYGEN_API_KEY"] = args.api_key
# Verify API key is set
if not os.getenv("HEYGEN_API_KEY"):
print("ERROR: HeyGen API key not provided.")
print(
"Please set it using --api-key or the HEYGEN_API_KEY environment variable."
)
sys.exit(1)
mcp.run()
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/heygen_mcp/api_client.py:
--------------------------------------------------------------------------------
```python
"""HeyGen API client module for interacting with the HeyGen API."""
import importlib.metadata
from typing import Any, Dict, List, Optional
import httpx
from pydantic import BaseModel, Field, HttpUrl
#######################
# HeyGen API Models #
#######################
# Common base response model
class BaseHeyGenResponse(BaseModel):
error: Optional[str] = None
# Voice information models
class VoiceInfo(BaseModel):
voice_id: str
language: str
gender: str
name: str
preview_audio: HttpUrl
support_pause: bool
emotion_support: bool
support_interactive_avatar: bool
class VoicesData(BaseModel):
voices: List[VoiceInfo]
class VoicesResponse(BaseHeyGenResponse):
data: Optional[VoicesData] = None
# User quota models
class QuotaDetails(BaseModel):
api: int
streaming_avatar: int
streaming_avatar_instance_quota: int
seat: int
class RemainingQuota(BaseModel):
remaining_quota: int
details: QuotaDetails
class RemainingQuotaResponse(BaseHeyGenResponse):
data: Optional[RemainingQuota] = None
# Avatar group models
class AvatarGroup(BaseModel):
id: str
name: str
created_at: int
num_looks: int
preview_image: HttpUrl
group_type: str
train_status: Optional[str] = None
class AvatarGroupListData(BaseModel):
total_count: int
avatar_group_list: List[AvatarGroup]
class AvatarGroupListResponse(BaseHeyGenResponse):
data: Optional[AvatarGroupListData] = None
# Avatar models
class Avatar(BaseModel):
avatar_id: str
avatar_name: str
gender: str
preview_image_url: HttpUrl
preview_video_url: HttpUrl
premium: bool
type: Optional[str] = None
tags: Optional[List[str]] = None
default_voice_id: Optional[str] = None
class AvatarsInGroupData(BaseModel):
avatar_list: List[Avatar]
class AvatarsInGroupResponse(BaseHeyGenResponse):
data: Optional[AvatarsInGroupData] = None
# Video generation models
class Character(BaseModel):
type: str = "avatar"
avatar_id: str
avatar_style: str = "normal"
scale: float = 1.0
class Voice(BaseModel):
type: str = "text"
input_text: str
voice_id: str
class VideoInput(BaseModel):
character: Character
voice: Voice
class Dimension(BaseModel):
width: int = 1280
height: int = 720
class VideoGenerateRequest(BaseModel):
title: str = ""
video_inputs: List[VideoInput]
test: bool = False
callback_id: Optional[str] = None
dimension: Dimension = Field(default_factory=lambda: Dimension())
aspect_ratio: Optional[str] = None
caption: bool = False
class VideoGenerateResponse(BaseHeyGenResponse):
data: Optional[Dict[str, Any]] = None
# Video status models
class VideoStatusError(BaseModel):
code: Optional[int] = None
detail: Optional[str] = None
message: Optional[str] = None
class VideoStatusData(BaseModel):
callback_id: Optional[str] = None
caption_url: Optional[str] = None
created_at: Optional[int] = None
duration: Optional[float] = None
error: Optional[VideoStatusError] = None
gif_url: Optional[str] = None
id: str
status: str # Values: "waiting", "pending", "processing", "completed", "failed"
thumbnail_url: Optional[str] = None
video_url: Optional[str] = None
video_url_caption: Optional[str] = None
class VideoStatusResponse(BaseModel):
code: int
data: VideoStatusData
message: str
########################
# MCP Response Models #
########################
class MCPGetCreditsResponse(BaseHeyGenResponse):
remaining_credits: Optional[int] = None
class MCPVoicesResponse(BaseHeyGenResponse):
voices: Optional[List[VoiceInfo]] = None
class MCPAvatarGroupResponse(BaseHeyGenResponse):
avatar_groups: Optional[List[AvatarGroup]] = None
total_count: Optional[int] = None
class MCPAvatarsInGroupResponse(BaseHeyGenResponse):
avatars: Optional[List[Avatar]] = None
class MCPVideoGenerateResponse(BaseHeyGenResponse):
video_id: Optional[str] = None
task_id: Optional[str] = None
video_url: Optional[str] = None
status: Optional[str] = None
class MCPVideoStatusResponse(BaseHeyGenResponse):
video_id: Optional[str] = None
status: Optional[str] = None
duration: Optional[float] = None
video_url: Optional[str] = None
gif_url: Optional[str] = None
thumbnail_url: Optional[str] = None
created_at: Optional[int] = None
error_details: Optional[Dict[str, Any]] = None
# HeyGen API Client Class
class HeyGenApiClient:
"""Client for interacting with the HeyGen API."""
def __init__(self, api_key: str):
"""Initialize the API client with the API key."""
self.api_key = api_key
# Set version for user agent
try:
self.version = importlib.metadata.version("heygen-mcp")
except importlib.metadata.PackageNotFoundError:
self.version = "unknown"
self.user_agent = f"heygen-mcp/{self.version}"
self.base_url = "https://api.heygen.com/v2"
self._client = httpx.AsyncClient()
async def close(self):
"""Close the underlying HTTP client."""
await self._client.aclose()
def _get_headers(self) -> Dict[str, str]:
"""Return the headers needed for API requests."""
return {
"Accept": "application/json",
"X-Api-Key": self.api_key,
"User-Agent": self.user_agent,
}
async def _make_request(
self, endpoint: str, method: str = "GET", data: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Make a request to the specified API endpoint.
Args:
endpoint: The API endpoint to call (without the base URL)
method: HTTP method to use (GET or POST)
data: JSON payload for POST requests
Returns:
The JSON response from the API
Raises:
httpx.RequestError: If there's a network-related error
httpx.HTTPStatusError: If the API returns an error status code
Exception: For any other unexpected errors
"""
url = f"{self.base_url}/{endpoint}"
headers = self._get_headers()
if method.upper() == "GET":
response = await self._client.get(url, headers=headers)
elif method.upper() == "POST":
headers["Content-Type"] = "application/json"
response = await self._client.post(url, headers=headers, json=data)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
response.raise_for_status() # Raises if status code is 4xx or 5xx
return response.json()
async def _handle_api_request(
self,
api_call,
response_model_class,
mcp_response_class,
error_msg: str,
**kwargs,
):
"""Generic handler for API requests to reduce code duplication.
Args:
api_call: Async function to call the API
response_model_class: Pydantic model class for validating the API response
mcp_response_class: Pydantic model class for the MCP response
error_msg: Error message to return if the validation fails
**kwargs: Additional arguments for the response transformation
Returns:
An MCP response object
"""
try:
# Make the request to the API
result = await api_call()
# Validate the response
validated_response = response_model_class.model_validate(result)
# Return the appropriate response based on the validation result
if hasattr(validated_response, "data") and validated_response.data:
return self._transform_to_mcp_response(
validated_response.data, mcp_response_class, **kwargs
)
elif validated_response.error:
return mcp_response_class(error=validated_response.error)
else:
return mcp_response_class(error=error_msg)
except httpx.RequestError as exc:
return mcp_response_class(error=f"HTTP Request failed: {exc}")
except httpx.HTTPStatusError as exc:
return mcp_response_class(
error=f"HTTP Error: {exc.response.status_code} - {exc.response.text}"
)
except Exception as e:
return mcp_response_class(error=f"An unexpected error occurred: {e}")
def _transform_to_mcp_response(self, data, mcp_response_class, **kwargs):
"""Transform API response data to MCP response format.
Args:
data: The API response data
mcp_response_class: The MCP response class to instantiate
**kwargs: Additional parameters for the response
Returns:
An instance of the MCP response class
"""
if "transform_func" in kwargs:
# Use the provided transform function
transform_func = kwargs.pop("transform_func")
return transform_func(data, mcp_response_class)
# Apply lambda functions to data if provided or use direct values
processed_kwargs = {}
for key, value in kwargs.items():
if callable(value):
processed_kwargs[key] = value(data)
else:
processed_kwargs[key] = value
return mcp_response_class(**processed_kwargs)
async def get_remaining_credits(self) -> MCPGetCreditsResponse:
"""Get the remaining credits from the API."""
async def api_call():
return await self._make_request("user/remaining_quota")
def transform_data(data, mcp_class):
return mcp_class(remaining_credits=int(data.remaining_quota / 60))
return await self._handle_api_request(
api_call=api_call,
response_model_class=RemainingQuotaResponse,
mcp_response_class=MCPGetCreditsResponse,
error_msg="No quota information found.",
transform_func=transform_data,
)
async def get_voices(self) -> MCPVoicesResponse:
"""Get the list of available voices from the API."""
async def api_call():
return await self._make_request("voices")
def transform_data(data, mcp_class):
# Truncate to the first 100 voices
return mcp_class(voices=data.voices[:100] if data.voices else None)
return await self._handle_api_request(
api_call=api_call,
response_model_class=VoicesResponse,
mcp_response_class=MCPVoicesResponse,
error_msg="No voices found.",
transform_func=transform_data,
)
async def list_avatar_groups(
self, include_public: bool = False
) -> MCPAvatarGroupResponse:
"""Get the list of avatar groups from the API."""
async def api_call():
public_param = "true" if include_public else "false"
endpoint = f"avatar_group.list?include_public={public_param}"
return await self._make_request(endpoint)
def transform_data(data, mcp_class):
return mcp_class(
avatar_groups=data.avatar_group_list, total_count=data.total_count
)
return await self._handle_api_request(
api_call=api_call,
response_model_class=AvatarGroupListResponse,
mcp_response_class=MCPAvatarGroupResponse,
error_msg="No avatar groups found.",
transform_func=transform_data,
)
async def get_avatars_in_group(self, group_id: str) -> MCPAvatarsInGroupResponse:
"""Get the list of avatars in a specific avatar group."""
async def api_call():
endpoint = f"avatar_group/{group_id}/avatars"
return await self._make_request(endpoint)
def transform_data(data, mcp_class):
return mcp_class(avatars=data.avatar_list)
return await self._handle_api_request(
api_call=api_call,
response_model_class=AvatarsInGroupResponse,
mcp_response_class=MCPAvatarsInGroupResponse,
error_msg="No avatars found in the group.",
transform_func=transform_data,
)
async def generate_avatar_video(
self, video_request: VideoGenerateRequest
) -> MCPVideoGenerateResponse:
"""Generate an avatar video using the HeyGen API."""
async def api_call():
return await self._make_request(
"video/generate", method="POST", data=video_request.model_dump()
)
return await self._handle_api_request(
api_call=api_call,
response_model_class=VideoGenerateResponse,
mcp_response_class=MCPVideoGenerateResponse,
error_msg="No video generation data returned.",
video_id=lambda d: d.get("video_id"),
task_id=lambda d: d.get("task_id"),
video_url=lambda d: d.get("video_url"),
status=lambda d: d.get("status"),
)
async def get_video_status(self, video_id: str) -> MCPVideoStatusResponse:
"""Get the status of a generated video from the API."""
async def api_call():
# The endpoint is v1, not v2
endpoint = f"../v1/video_status.get?video_id={video_id}"
return await self._make_request(endpoint)
try:
# Make the request to the API
result = await api_call()
# Validate the response
validated_response = VideoStatusResponse.model_validate(result)
# Extract data
data = validated_response.data
# Process error details if present
error_details = None
if data.error:
error_details = {
"code": data.error.code,
"message": data.error.message,
"detail": data.error.detail,
}
# Return MCP response
return MCPVideoStatusResponse(
video_id=data.id,
status=data.status,
duration=data.duration,
video_url=data.video_url,
gif_url=data.gif_url,
thumbnail_url=data.thumbnail_url,
created_at=data.created_at,
error_details=error_details,
)
except httpx.RequestError as exc:
return MCPVideoStatusResponse(error=f"HTTP Request failed: {exc}")
except httpx.HTTPStatusError as exc:
return MCPVideoStatusResponse(
error=f"HTTP Error: {exc.response.status_code} - {exc.response.text}"
)
except Exception as e:
return MCPVideoStatusResponse(error=f"An unexpected error occurred: {e}")
```