# 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:
--------------------------------------------------------------------------------
```
1 | 3.12
2 |
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
1 | HEYGEN_API_KEY="<API_KEY_HERE>"
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | # Python
2 | __pycache__/
3 | *.py[cod]
4 | *$py.class
5 | *.so
6 | .Python
7 | build/
8 | develop-eggs/
9 | dist/
10 | downloads/
11 | eggs/
12 | .eggs/
13 | lib/
14 | lib64/
15 | parts/
16 | sdist/
17 | var/
18 | wheels/
19 | *.egg-info/
20 | .installed.cfg
21 | *.egg
22 |
23 | # Environment
24 | .env
25 | .venv
26 | env/
27 | venv/
28 | ENV/
29 | env.bak/
30 | venv.bak/
31 |
32 | # IDE
33 | .idea/
34 | .vscode/
35 | *.swp
36 | *.swo
37 |
38 | # Logs
39 | *.log
40 |
41 | # Local development
42 | .DS_Store
43 | .ruff_cache/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Heygen MCP Server
2 |
3 | 
4 |
5 | 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.
6 |
7 | [](https://opensource.org/licenses/MIT)
8 |
9 | Note: This project is in early development. While we welcome community feedback and contributions, please be aware that official support is limited.
10 |
11 | ## Installation
12 |
13 | ### Prerequisites
14 |
15 | - Python 3.10 or higher
16 | - A Heygen API key (get one from [Heygen](https://www.heygen.com/)). Includes 10 Free Credits per Month
17 |
18 | ### Installing uv
19 |
20 | uv is a fast Python package installer and resolver that we recommend for installing this package.
21 |
22 | **macOS or Linux:**
23 |
24 | ```bash
25 | # Install with the official installer script
26 | curl -LsSf https://astral.sh/uv/install.sh | sh
27 |
28 | # Or via Homebrew (macOS)
29 | brew install uv
30 | ```
31 |
32 | **Windows:**
33 |
34 | ```powershell
35 | # Install with the official installer script in PowerShell
36 | irm https://astral.sh/uv/install.ps1 | iex
37 |
38 | # Or via Scoop
39 | scoop install uv
40 | ```
41 |
42 | For other installation methods, see the [uv documentation](https://github.com/astral-sh/uv).
43 |
44 | ## Usage
45 |
46 | ### Quickstart with Claude Desktop
47 |
48 | 1. Get your API key from [HeyGen](https://www.heygen.com/).
49 | 2. Install uv package manager (see [Installing uv](#installing-uv) section above).
50 | 3. Go to Claude > Settings > Developer > Edit Config > `claude_desktop_config.json` to include the following:
51 |
52 | ```json
53 | {
54 | "mcpServers": {
55 | "HeyGen": {
56 | "command": "uvx",
57 | "args": ["heygen-mcp"],
58 | "env": {
59 | "HEYGEN_API_KEY": "<insert-your-api-key-here>"
60 | }
61 | }
62 | }
63 | }
64 | ```
65 |
66 | 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".
67 |
68 | ### Available MCP Tools
69 |
70 | The server provides the following tools to Claude:
71 |
72 | - **get_remaining_credits**: Retrieves the remaining credits in your Heygen account.
73 | - **get_voices**: Retrieves a list of available voices from the Heygen API (limited to first 100 voices).
74 | - **get_avatar_groups**: Retrieves a list of Heygen avatar groups.
75 | - **get_avatars_in_avatar_group**: Retrieves a list of avatars in a specific Heygen avatar group.
76 | - **generate_avatar_video**: Generates a new avatar video with the specified avatar, text, and voice.
77 | - **get_avatar_video_status**: Retrieves the status of a video generated via the Heygen API.
78 |
79 | ## Development
80 |
81 | ### Running with MCP Inspector
82 |
83 | To run the server locally with the MCP Inspector for testing and debugging:
84 |
85 | ```bash
86 | uv --with "mcp[cli]" dev heygen_mcp/server.py
87 | ```
88 |
89 | This will start the server in development mode and allow you to use the MCP Inspector to test the available tools and functionality.
90 |
91 | ## Roadmap
92 |
93 | - [ ] Tests
94 | - [ ] CICD
95 | - [ ] Photo Avatar APIs Support
96 | - [ ] SSE And Remote MCP Server with OAuth Flow
97 | - [ ] Translation API Support
98 | - [ ] Template API Support
99 | - [ ] Interactive Avatar API Support
100 |
101 | ## Contributing
102 |
103 | Contributions are welcome! Please feel free to submit a Pull Request.
104 |
105 | ## License
106 |
107 | This project is licensed under the MIT License - see the LICENSE file for details.
108 |
```
--------------------------------------------------------------------------------
/build_and_publish.sh:
--------------------------------------------------------------------------------
```bash
1 | #!/bin/bash
2 |
3 | # Remove dist folder if it exists
4 | rm -rf dist
5 | rm -rf *.egg-info
6 |
7 | # Build package
8 | uv build
9 |
10 | # Publish to PyPI
11 | uv publish
12 |
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python
2 | """Setup script for backwards compatibility with older pip versions."""
3 |
4 | from setuptools import setup
5 |
6 | if __name__ == "__main__":
7 | setup()
8 |
```
--------------------------------------------------------------------------------
/run.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python
2 | """Simple entry point to run the heygen-mcp server during development."""
3 |
4 | from heygen_mcp.server import main
5 |
6 | if __name__ == "__main__":
7 | main()
8 |
```
--------------------------------------------------------------------------------
/heygen_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """HeyGen MCP - API client and MCP server for HeyGen API interaction."""
2 |
3 | __version__ = "0.0.3"
4 |
5 | from heygen_mcp.api_client import HeyGenApiClient
6 | from heygen_mcp.server import main, mcp
7 |
8 | __all__ = ["HeyGenApiClient", "mcp", "main"]
9 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [build-system]
2 | requires = ["setuptools>=61.0", "wheel"]
3 | build-backend = "setuptools.build_meta"
4 |
5 | [project]
6 | name = "heygen-mcp"
7 | version = "0.0.3"
8 | description = "HeyGen MCP Server for AI Video Creation"
9 | readme = "README.md"
10 | requires-python = ">=3.12"
11 | license = {text = "MIT"}
12 | authors = [
13 | { name = "Eddy Kim", email = "[email protected]" },
14 | ]
15 | keywords = ["heygen", "mcp", "claude", "ai", "video", "avatar"]
16 | classifiers = [
17 | "Development Status :: 3 - Alpha",
18 | "Intended Audience :: Developers",
19 | "License :: OSI Approved :: MIT License",
20 | "Programming Language :: Python :: 3",
21 | "Programming Language :: Python :: 3.12",
22 | ]
23 |
24 | dependencies = [
25 | "mcp[cli]>=1.6.0",
26 | "pydantic>=2.0.0",
27 | "httpx>=0.27.0",
28 | "python-dotenv>=1.0.0",
29 | ]
30 |
31 | [project.scripts]
32 | heygen-mcp = "heygen_mcp.server:main"
33 |
34 | [project.urls]
35 | "Homepage" = "https://github.com/heygen-com/heygen-mcp"
36 | "Bug Tracker" = "https://github.com/heygen-com/heygen-mcp/issues"
37 |
38 | [project.optional-dependencies]
39 | dev = [
40 | "pytest",
41 | "pytest-asyncio",
42 | "ruff",
43 | "build",
44 | "twine",
45 | ]
46 |
47 | [tool.ruff]
48 | line-length = 88
49 | target-version = "py312"
50 |
51 | [tool.ruff.lint]
52 | select = ["E", "F", "B", "I"]
53 |
54 | [tool.ruff.format]
55 | quote-style = "double"
56 | indent-style = "space"
57 |
58 | [tool.uv]
59 | # The commands subsection is not valid in tool.uv
60 | # The entry point is already defined in [project.scripts]
61 |
```
--------------------------------------------------------------------------------
/heygen_mcp/server.py:
--------------------------------------------------------------------------------
```python
1 | """HeyGen MCP server module for providing MCP tools for the HeyGen API."""
2 |
3 | import argparse
4 | import os
5 | import sys
6 |
7 | from dotenv import load_dotenv
8 | from mcp.server.fastmcp import FastMCP
9 |
10 | from heygen_mcp.api_client import (
11 | Character,
12 | Dimension,
13 | HeyGenApiClient,
14 | MCPAvatarGroupResponse,
15 | MCPAvatarsInGroupResponse,
16 | MCPGetCreditsResponse,
17 | MCPVideoGenerateResponse,
18 | MCPVideoStatusResponse,
19 | MCPVoicesResponse,
20 | VideoGenerateRequest,
21 | VideoInput,
22 | Voice,
23 | )
24 |
25 | # Load environment variables
26 | load_dotenv()
27 |
28 | # Create MCP server instance
29 | mcp = FastMCP("HeyGen MCP")
30 | api_client = None
31 |
32 |
33 | # Function to get or create API client
34 | async def get_api_client() -> HeyGenApiClient:
35 | """Get the API client, creating it if necessary."""
36 | global api_client
37 |
38 | # If we already have a client, return it
39 | if api_client is not None:
40 | return api_client
41 |
42 | # Otherwise, get the API key and create a new client
43 | api_key = os.getenv("HEYGEN_API_KEY")
44 |
45 | if not api_key:
46 | raise ValueError("HEYGEN_API_KEY environment variable not set.")
47 |
48 | # Create and store the client
49 | api_client = HeyGenApiClient(api_key)
50 | return api_client
51 |
52 |
53 | ########################
54 | # MCP Tool Definitions #
55 | ########################
56 |
57 |
58 | @mcp.tool(
59 | name="get_remaining_credits",
60 | description="Retrieves the remaining credits in heygen account.",
61 | )
62 | async def get_remaining_credits() -> MCPGetCreditsResponse:
63 | """Get the remaining quota for the user via HeyGen API."""
64 | try:
65 | client = await get_api_client()
66 | return await client.get_remaining_credits()
67 | except Exception as e:
68 | return MCPGetCreditsResponse(error=str(e))
69 |
70 |
71 | @mcp.tool(
72 | name="get_voices",
73 | description=(
74 | "Retrieves a list of available voices from the HeyGen API. Results truncated "
75 | "to first 100 voices. Private voices generally will returned 1st."
76 | ),
77 | )
78 | async def get_voices() -> MCPVoicesResponse:
79 | """Get the list of available voices via HeyGen API."""
80 | try:
81 | client = await get_api_client()
82 | return await client.get_voices()
83 | except Exception as e:
84 | return MCPVoicesResponse(error=str(e))
85 |
86 |
87 | @mcp.tool(
88 | name="get_avatar_groups",
89 | description=(
90 | "Retrieves a list of HeyGen avatar groups. By default, only private avatar "
91 | "groups are returned, unless include_public is set to true. Avatar groups "
92 | "are collections of avatars, avatar group ids cannot be used to generate "
93 | "videos."
94 | ),
95 | )
96 | async def get_avatar_groups(include_public: bool = False) -> MCPAvatarGroupResponse:
97 | """List avatar groups via HeyGen API v2/avatar_group.list endpoint."""
98 | try:
99 | client = await get_api_client()
100 | return await client.list_avatar_groups(include_public)
101 | except Exception as e:
102 | return MCPAvatarGroupResponse(error=str(e))
103 |
104 |
105 | @mcp.tool(
106 | name="get_avatars_in_avatar_group",
107 | description="Retrieves a list of avatars in a specific HeyGen avatar group.",
108 | )
109 | async def get_avatars_in_avatar_group(group_id: str) -> MCPAvatarsInGroupResponse:
110 | """List avatars in a specific HeyGen avatar group via HeyGen API."""
111 | try:
112 | client = await get_api_client()
113 | return await client.get_avatars_in_group(group_id)
114 | except Exception as e:
115 | return MCPAvatarsInGroupResponse(error=str(e))
116 |
117 |
118 | @mcp.tool(
119 | name="generate_avatar_video",
120 | description="Generates a new avatar video via the HeyGen API.",
121 | )
122 | async def generate_avatar_video(
123 | avatar_id: str, input_text: str, voice_id: str, title: str = ""
124 | ) -> MCPVideoGenerateResponse:
125 | """Generate a new avatar video using the HeyGen API."""
126 | try:
127 | # Create the request object with default values
128 | request = VideoGenerateRequest(
129 | title=title,
130 | video_inputs=[
131 | VideoInput(
132 | character=Character(avatar_id=avatar_id),
133 | voice=Voice(input_text=input_text, voice_id=voice_id),
134 | )
135 | ],
136 | dimension=Dimension(width=1280, height=720),
137 | )
138 |
139 | client = await get_api_client()
140 | return await client.generate_avatar_video(request)
141 | except Exception as e:
142 | return MCPVideoGenerateResponse(error=str(e))
143 |
144 |
145 | @mcp.tool(
146 | name="get_avatar_video_status",
147 | description=(
148 | "Retrieves the status of a video generated via the HeyGen API. Video status "
149 | "make take several minutes to hours depending on length of video and queue "
150 | "time. If video is not yet complete, status be viewed later by user via "
151 | "https://app.heygen.com/home"
152 | ),
153 | )
154 | async def get_avatar_video_status(video_id: str) -> MCPVideoStatusResponse:
155 | """Retrieve the status of a video generated via the HeyGen API."""
156 | try:
157 | client = await get_api_client()
158 | return await client.get_video_status(video_id)
159 | except Exception as e:
160 | return MCPVideoStatusResponse(error=str(e))
161 |
162 |
163 | def parse_args():
164 | """Parse command-line arguments."""
165 | parser = argparse.ArgumentParser(description="HeyGen MCP Server")
166 | parser.add_argument(
167 | "--api-key",
168 | help=(
169 | "HeyGen API key. Alternatively, set HEYGEN_API_KEY environment variable."
170 | ),
171 | )
172 | parser.add_argument(
173 | "--host", default="127.0.0.1", help="Host to bind the server to."
174 | )
175 | parser.add_argument(
176 | "--port", type=int, default=8000, help="Port to bind the server to."
177 | )
178 | parser.add_argument(
179 | "--reload",
180 | action="store_true",
181 | help="Enable auto-reload for development.",
182 | )
183 | return parser.parse_args()
184 |
185 |
186 | def main():
187 | """Run the MCP server."""
188 | args = parse_args()
189 |
190 | # Check if API key is provided or in environment
191 | if args.api_key:
192 | os.environ["HEYGEN_API_KEY"] = args.api_key
193 |
194 | # Verify API key is set
195 | if not os.getenv("HEYGEN_API_KEY"):
196 | print("ERROR: HeyGen API key not provided.")
197 | print(
198 | "Please set it using --api-key or the HEYGEN_API_KEY environment variable."
199 | )
200 | sys.exit(1)
201 |
202 | mcp.run()
203 |
204 |
205 | if __name__ == "__main__":
206 | main()
207 |
```
--------------------------------------------------------------------------------
/heygen_mcp/api_client.py:
--------------------------------------------------------------------------------
```python
1 | """HeyGen API client module for interacting with the HeyGen API."""
2 |
3 | import importlib.metadata
4 | from typing import Any, Dict, List, Optional
5 |
6 | import httpx
7 | from pydantic import BaseModel, Field, HttpUrl
8 |
9 | #######################
10 | # HeyGen API Models #
11 | #######################
12 |
13 |
14 | # Common base response model
15 | class BaseHeyGenResponse(BaseModel):
16 | error: Optional[str] = None
17 |
18 |
19 | # Voice information models
20 | class VoiceInfo(BaseModel):
21 | voice_id: str
22 | language: str
23 | gender: str
24 | name: str
25 | preview_audio: HttpUrl
26 | support_pause: bool
27 | emotion_support: bool
28 | support_interactive_avatar: bool
29 |
30 |
31 | class VoicesData(BaseModel):
32 | voices: List[VoiceInfo]
33 |
34 |
35 | class VoicesResponse(BaseHeyGenResponse):
36 | data: Optional[VoicesData] = None
37 |
38 |
39 | # User quota models
40 | class QuotaDetails(BaseModel):
41 | api: int
42 | streaming_avatar: int
43 | streaming_avatar_instance_quota: int
44 | seat: int
45 |
46 |
47 | class RemainingQuota(BaseModel):
48 | remaining_quota: int
49 | details: QuotaDetails
50 |
51 |
52 | class RemainingQuotaResponse(BaseHeyGenResponse):
53 | data: Optional[RemainingQuota] = None
54 |
55 |
56 | # Avatar group models
57 | class AvatarGroup(BaseModel):
58 | id: str
59 | name: str
60 | created_at: int
61 | num_looks: int
62 | preview_image: HttpUrl
63 | group_type: str
64 | train_status: Optional[str] = None
65 |
66 |
67 | class AvatarGroupListData(BaseModel):
68 | total_count: int
69 | avatar_group_list: List[AvatarGroup]
70 |
71 |
72 | class AvatarGroupListResponse(BaseHeyGenResponse):
73 | data: Optional[AvatarGroupListData] = None
74 |
75 |
76 | # Avatar models
77 | class Avatar(BaseModel):
78 | avatar_id: str
79 | avatar_name: str
80 | gender: str
81 | preview_image_url: HttpUrl
82 | preview_video_url: HttpUrl
83 | premium: bool
84 | type: Optional[str] = None
85 | tags: Optional[List[str]] = None
86 | default_voice_id: Optional[str] = None
87 |
88 |
89 | class AvatarsInGroupData(BaseModel):
90 | avatar_list: List[Avatar]
91 |
92 |
93 | class AvatarsInGroupResponse(BaseHeyGenResponse):
94 | data: Optional[AvatarsInGroupData] = None
95 |
96 |
97 | # Video generation models
98 | class Character(BaseModel):
99 | type: str = "avatar"
100 | avatar_id: str
101 | avatar_style: str = "normal"
102 | scale: float = 1.0
103 |
104 |
105 | class Voice(BaseModel):
106 | type: str = "text"
107 | input_text: str
108 | voice_id: str
109 |
110 |
111 | class VideoInput(BaseModel):
112 | character: Character
113 | voice: Voice
114 |
115 |
116 | class Dimension(BaseModel):
117 | width: int = 1280
118 | height: int = 720
119 |
120 |
121 | class VideoGenerateRequest(BaseModel):
122 | title: str = ""
123 | video_inputs: List[VideoInput]
124 | test: bool = False
125 | callback_id: Optional[str] = None
126 | dimension: Dimension = Field(default_factory=lambda: Dimension())
127 | aspect_ratio: Optional[str] = None
128 | caption: bool = False
129 |
130 |
131 | class VideoGenerateResponse(BaseHeyGenResponse):
132 | data: Optional[Dict[str, Any]] = None
133 |
134 |
135 | # Video status models
136 | class VideoStatusError(BaseModel):
137 | code: Optional[int] = None
138 | detail: Optional[str] = None
139 | message: Optional[str] = None
140 |
141 |
142 | class VideoStatusData(BaseModel):
143 | callback_id: Optional[str] = None
144 | caption_url: Optional[str] = None
145 | created_at: Optional[int] = None
146 | duration: Optional[float] = None
147 | error: Optional[VideoStatusError] = None
148 | gif_url: Optional[str] = None
149 | id: str
150 | status: str # Values: "waiting", "pending", "processing", "completed", "failed"
151 | thumbnail_url: Optional[str] = None
152 | video_url: Optional[str] = None
153 | video_url_caption: Optional[str] = None
154 |
155 |
156 | class VideoStatusResponse(BaseModel):
157 | code: int
158 | data: VideoStatusData
159 | message: str
160 |
161 |
162 | ########################
163 | # MCP Response Models #
164 | ########################
165 |
166 |
167 | class MCPGetCreditsResponse(BaseHeyGenResponse):
168 | remaining_credits: Optional[int] = None
169 |
170 |
171 | class MCPVoicesResponse(BaseHeyGenResponse):
172 | voices: Optional[List[VoiceInfo]] = None
173 |
174 |
175 | class MCPAvatarGroupResponse(BaseHeyGenResponse):
176 | avatar_groups: Optional[List[AvatarGroup]] = None
177 | total_count: Optional[int] = None
178 |
179 |
180 | class MCPAvatarsInGroupResponse(BaseHeyGenResponse):
181 | avatars: Optional[List[Avatar]] = None
182 |
183 |
184 | class MCPVideoGenerateResponse(BaseHeyGenResponse):
185 | video_id: Optional[str] = None
186 | task_id: Optional[str] = None
187 | video_url: Optional[str] = None
188 | status: Optional[str] = None
189 |
190 |
191 | class MCPVideoStatusResponse(BaseHeyGenResponse):
192 | video_id: Optional[str] = None
193 | status: Optional[str] = None
194 | duration: Optional[float] = None
195 | video_url: Optional[str] = None
196 | gif_url: Optional[str] = None
197 | thumbnail_url: Optional[str] = None
198 | created_at: Optional[int] = None
199 | error_details: Optional[Dict[str, Any]] = None
200 |
201 |
202 | # HeyGen API Client Class
203 | class HeyGenApiClient:
204 | """Client for interacting with the HeyGen API."""
205 |
206 | def __init__(self, api_key: str):
207 | """Initialize the API client with the API key."""
208 | self.api_key = api_key
209 |
210 | # Set version for user agent
211 | try:
212 | self.version = importlib.metadata.version("heygen-mcp")
213 | except importlib.metadata.PackageNotFoundError:
214 | self.version = "unknown"
215 |
216 | self.user_agent = f"heygen-mcp/{self.version}"
217 | self.base_url = "https://api.heygen.com/v2"
218 | self._client = httpx.AsyncClient()
219 |
220 | async def close(self):
221 | """Close the underlying HTTP client."""
222 | await self._client.aclose()
223 |
224 | def _get_headers(self) -> Dict[str, str]:
225 | """Return the headers needed for API requests."""
226 | return {
227 | "Accept": "application/json",
228 | "X-Api-Key": self.api_key,
229 | "User-Agent": self.user_agent,
230 | }
231 |
232 | async def _make_request(
233 | self, endpoint: str, method: str = "GET", data: Optional[Dict[str, Any]] = None
234 | ) -> Dict[str, Any]:
235 | """Make a request to the specified API endpoint.
236 |
237 | Args:
238 | endpoint: The API endpoint to call (without the base URL)
239 | method: HTTP method to use (GET or POST)
240 | data: JSON payload for POST requests
241 |
242 | Returns:
243 | The JSON response from the API
244 |
245 | Raises:
246 | httpx.RequestError: If there's a network-related error
247 | httpx.HTTPStatusError: If the API returns an error status code
248 | Exception: For any other unexpected errors
249 | """
250 | url = f"{self.base_url}/{endpoint}"
251 | headers = self._get_headers()
252 |
253 | if method.upper() == "GET":
254 | response = await self._client.get(url, headers=headers)
255 | elif method.upper() == "POST":
256 | headers["Content-Type"] = "application/json"
257 | response = await self._client.post(url, headers=headers, json=data)
258 | else:
259 | raise ValueError(f"Unsupported HTTP method: {method}")
260 |
261 | response.raise_for_status() # Raises if status code is 4xx or 5xx
262 | return response.json()
263 |
264 | async def _handle_api_request(
265 | self,
266 | api_call,
267 | response_model_class,
268 | mcp_response_class,
269 | error_msg: str,
270 | **kwargs,
271 | ):
272 | """Generic handler for API requests to reduce code duplication.
273 |
274 | Args:
275 | api_call: Async function to call the API
276 | response_model_class: Pydantic model class for validating the API response
277 | mcp_response_class: Pydantic model class for the MCP response
278 | error_msg: Error message to return if the validation fails
279 | **kwargs: Additional arguments for the response transformation
280 |
281 | Returns:
282 | An MCP response object
283 | """
284 | try:
285 | # Make the request to the API
286 | result = await api_call()
287 |
288 | # Validate the response
289 | validated_response = response_model_class.model_validate(result)
290 |
291 | # Return the appropriate response based on the validation result
292 | if hasattr(validated_response, "data") and validated_response.data:
293 | return self._transform_to_mcp_response(
294 | validated_response.data, mcp_response_class, **kwargs
295 | )
296 | elif validated_response.error:
297 | return mcp_response_class(error=validated_response.error)
298 | else:
299 | return mcp_response_class(error=error_msg)
300 |
301 | except httpx.RequestError as exc:
302 | return mcp_response_class(error=f"HTTP Request failed: {exc}")
303 | except httpx.HTTPStatusError as exc:
304 | return mcp_response_class(
305 | error=f"HTTP Error: {exc.response.status_code} - {exc.response.text}"
306 | )
307 | except Exception as e:
308 | return mcp_response_class(error=f"An unexpected error occurred: {e}")
309 |
310 | def _transform_to_mcp_response(self, data, mcp_response_class, **kwargs):
311 | """Transform API response data to MCP response format.
312 |
313 | Args:
314 | data: The API response data
315 | mcp_response_class: The MCP response class to instantiate
316 | **kwargs: Additional parameters for the response
317 |
318 | Returns:
319 | An instance of the MCP response class
320 | """
321 | if "transform_func" in kwargs:
322 | # Use the provided transform function
323 | transform_func = kwargs.pop("transform_func")
324 | return transform_func(data, mcp_response_class)
325 |
326 | # Apply lambda functions to data if provided or use direct values
327 | processed_kwargs = {}
328 | for key, value in kwargs.items():
329 | if callable(value):
330 | processed_kwargs[key] = value(data)
331 | else:
332 | processed_kwargs[key] = value
333 |
334 | return mcp_response_class(**processed_kwargs)
335 |
336 | async def get_remaining_credits(self) -> MCPGetCreditsResponse:
337 | """Get the remaining credits from the API."""
338 |
339 | async def api_call():
340 | return await self._make_request("user/remaining_quota")
341 |
342 | def transform_data(data, mcp_class):
343 | return mcp_class(remaining_credits=int(data.remaining_quota / 60))
344 |
345 | return await self._handle_api_request(
346 | api_call=api_call,
347 | response_model_class=RemainingQuotaResponse,
348 | mcp_response_class=MCPGetCreditsResponse,
349 | error_msg="No quota information found.",
350 | transform_func=transform_data,
351 | )
352 |
353 | async def get_voices(self) -> MCPVoicesResponse:
354 | """Get the list of available voices from the API."""
355 |
356 | async def api_call():
357 | return await self._make_request("voices")
358 |
359 | def transform_data(data, mcp_class):
360 | # Truncate to the first 100 voices
361 | return mcp_class(voices=data.voices[:100] if data.voices else None)
362 |
363 | return await self._handle_api_request(
364 | api_call=api_call,
365 | response_model_class=VoicesResponse,
366 | mcp_response_class=MCPVoicesResponse,
367 | error_msg="No voices found.",
368 | transform_func=transform_data,
369 | )
370 |
371 | async def list_avatar_groups(
372 | self, include_public: bool = False
373 | ) -> MCPAvatarGroupResponse:
374 | """Get the list of avatar groups from the API."""
375 |
376 | async def api_call():
377 | public_param = "true" if include_public else "false"
378 | endpoint = f"avatar_group.list?include_public={public_param}"
379 | return await self._make_request(endpoint)
380 |
381 | def transform_data(data, mcp_class):
382 | return mcp_class(
383 | avatar_groups=data.avatar_group_list, total_count=data.total_count
384 | )
385 |
386 | return await self._handle_api_request(
387 | api_call=api_call,
388 | response_model_class=AvatarGroupListResponse,
389 | mcp_response_class=MCPAvatarGroupResponse,
390 | error_msg="No avatar groups found.",
391 | transform_func=transform_data,
392 | )
393 |
394 | async def get_avatars_in_group(self, group_id: str) -> MCPAvatarsInGroupResponse:
395 | """Get the list of avatars in a specific avatar group."""
396 |
397 | async def api_call():
398 | endpoint = f"avatar_group/{group_id}/avatars"
399 | return await self._make_request(endpoint)
400 |
401 | def transform_data(data, mcp_class):
402 | return mcp_class(avatars=data.avatar_list)
403 |
404 | return await self._handle_api_request(
405 | api_call=api_call,
406 | response_model_class=AvatarsInGroupResponse,
407 | mcp_response_class=MCPAvatarsInGroupResponse,
408 | error_msg="No avatars found in the group.",
409 | transform_func=transform_data,
410 | )
411 |
412 | async def generate_avatar_video(
413 | self, video_request: VideoGenerateRequest
414 | ) -> MCPVideoGenerateResponse:
415 | """Generate an avatar video using the HeyGen API."""
416 |
417 | async def api_call():
418 | return await self._make_request(
419 | "video/generate", method="POST", data=video_request.model_dump()
420 | )
421 |
422 | return await self._handle_api_request(
423 | api_call=api_call,
424 | response_model_class=VideoGenerateResponse,
425 | mcp_response_class=MCPVideoGenerateResponse,
426 | error_msg="No video generation data returned.",
427 | video_id=lambda d: d.get("video_id"),
428 | task_id=lambda d: d.get("task_id"),
429 | video_url=lambda d: d.get("video_url"),
430 | status=lambda d: d.get("status"),
431 | )
432 |
433 | async def get_video_status(self, video_id: str) -> MCPVideoStatusResponse:
434 | """Get the status of a generated video from the API."""
435 |
436 | async def api_call():
437 | # The endpoint is v1, not v2
438 | endpoint = f"../v1/video_status.get?video_id={video_id}"
439 | return await self._make_request(endpoint)
440 |
441 | try:
442 | # Make the request to the API
443 | result = await api_call()
444 |
445 | # Validate the response
446 | validated_response = VideoStatusResponse.model_validate(result)
447 |
448 | # Extract data
449 | data = validated_response.data
450 |
451 | # Process error details if present
452 | error_details = None
453 | if data.error:
454 | error_details = {
455 | "code": data.error.code,
456 | "message": data.error.message,
457 | "detail": data.error.detail,
458 | }
459 |
460 | # Return MCP response
461 | return MCPVideoStatusResponse(
462 | video_id=data.id,
463 | status=data.status,
464 | duration=data.duration,
465 | video_url=data.video_url,
466 | gif_url=data.gif_url,
467 | thumbnail_url=data.thumbnail_url,
468 | created_at=data.created_at,
469 | error_details=error_details,
470 | )
471 | except httpx.RequestError as exc:
472 | return MCPVideoStatusResponse(error=f"HTTP Request failed: {exc}")
473 | except httpx.HTTPStatusError as exc:
474 | return MCPVideoStatusResponse(
475 | error=f"HTTP Error: {exc.response.status_code} - {exc.response.text}"
476 | )
477 | except Exception as e:
478 | return MCPVideoStatusResponse(error=f"An unexpected error occurred: {e}")
479 |
```