# Directory Structure
```
├── api
│ ├── __init__.py
│ ├── config.py
│ ├── generate.py
│ ├── models.py
│ ├── storage.py
│ └── utils.py
├── LICENSE
├── main.py
└── README.md
```
# Files
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # fal.ai MCP Server
2 |
3 | A Model Context Protocol (MCP) server for interacting with fal.ai models and services.
4 |
5 | ## Features
6 |
7 | - List all available fal.ai models
8 | - Search for specific models by keywords
9 | - Get model schemas
10 | - Generate content using any fal.ai model
11 | - Support for both direct and queued model execution
12 | - Queue management (status checking, getting results, cancelling requests)
13 | - File upload to fal.ai CDN
14 |
15 | ## Requirements
16 |
17 | - Python 3.10+
18 | - fastmcp
19 | - httpx
20 | - aiofiles
21 | - A fal.ai API key
22 |
23 | ## Installation
24 |
25 | 1. Clone this repository:
26 | ```bash
27 | git clone https://github.com/am0y/mcp-fal.git
28 | cd mcp-fal
29 | ```
30 |
31 | 2. Install the required packages:
32 | ```bash
33 | pip install fastmcp httpx aiofiles
34 | ```
35 |
36 | 3. Set your fal.ai API key as an environment variable:
37 | ```bash
38 | export FAL_KEY="YOUR_FAL_API_KEY_HERE"
39 | ```
40 |
41 | ## Usage
42 |
43 | ### Running the Server
44 |
45 | You can run the server in development mode with:
46 |
47 | ```bash
48 | fastmcp dev main.py
49 | ```
50 |
51 | This will launch the MCP Inspector web interface where you can test the tools interactively.
52 |
53 | ### Installing in Claude Desktop
54 |
55 | To use the server with Claude Desktop:
56 |
57 | ```bash
58 | fastmcp install main.py -e FAL_KEY="YOUR_FAL_API_KEY_HERE"
59 | ```
60 |
61 | This will make the server available to Claude in the Desktop app.
62 |
63 | ### Running Directly
64 |
65 | You can also run the server directly:
66 |
67 | ```bash
68 | python main.py
69 | ```
70 |
71 | ## API Reference
72 |
73 | ### Tools
74 |
75 | - `models(page=None, total=None)` - List available models with optional pagination
76 | - `search(keywords)` - Search for models by keywords
77 | - `schema(model_id)` - Get OpenAPI schema for a specific model
78 | - `generate(model, parameters, queue=False)` - Generate content using a model
79 | - `result(url)` - Get result from a queued request
80 | - `status(url)` - Check status of a queued request
81 | - `cancel(url)` - Cancel a queued request
82 | - `upload(path` - Upload a file to fal.ai CDN
83 |
84 | ## License
85 |
86 | [MIT](LICENSE)
```
--------------------------------------------------------------------------------
/api/__init__.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | fal.ai MCP Server API package.
3 |
4 | This package contains modules for interacting with fal.ai models and services through the Model Context Protocol (MCP).
5 | """
6 |
7 | __version__ = "1.0.0"
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | fal.ai MCP Server : Main entry point
3 |
4 | This module sets up and runs the fal.ai MCP server,
5 | providing tools to interact with fal.ai models and services.
6 | """
7 |
8 | import os
9 | import sys
10 | from fastmcp import FastMCP
11 | from api.models import register_model_tools
12 | from api.generate import register_generation_tools
13 | from api.storage import register_storage_tools
14 | from api.config import get_api_key, SERVER_NAME, SERVER_DESCRIPTION, SERVER_VERSION, SERVER_DEPENDENCIES
15 |
16 | mcp = FastMCP(
17 | SERVER_NAME,
18 | description=SERVER_DESCRIPTION,
19 | dependencies=SERVER_DEPENDENCIES,
20 | version=SERVER_VERSION
21 | )
22 |
23 | register_model_tools(mcp)
24 | register_generation_tools(mcp)
25 | register_storage_tools(mcp)
26 |
27 | def main():
28 | try:
29 | get_api_key()
30 | except ValueError:
31 | pass
32 |
33 | try:
34 | mcp.run()
35 | except Exception as e:
36 | sys.exit(1)
37 |
38 | if __name__ == "__main__":
39 | main()
```
--------------------------------------------------------------------------------
/api/config.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Configuration module for the fal.ai MCP server.
3 |
4 | This module provides centralized configuration settings for the fal.ai MCP server, including API endpoints, timeouts, and more.
5 | """
6 |
7 | import os
8 | from typing import Dict, Any, Optional
9 |
10 | FAL_BASE_URL = "https://fal.ai/api"
11 | FAL_QUEUE_URL = "https://queue.fal.run"
12 | FAL_DIRECT_URL = "https://fal.run"
13 | FAL_REST_URL = "https://rest.alpha.fal.ai"
14 |
15 | DEFAULT_TIMEOUT = 30.0
16 | AUTHENTICATED_TIMEOUT = 100.0
17 |
18 | API_KEY_ENV_VAR = "FAL_KEY"
19 |
20 | SERVER_NAME = "fal.ai MCP Server"
21 | SERVER_DESCRIPTION = "Access fal.ai models and generate content through MCP"
22 | SERVER_VERSION = "1.0.0"
23 | SERVER_DEPENDENCIES = ["httpx", "aiofiles"]
24 |
25 |
26 | def get_env(key: str, default: Optional[str] = None) -> Optional[str]:
27 | """
28 | Get an environment variable with optional default value.
29 |
30 | Args:
31 | key: The name of the environment variable
32 | default: Optional default value if not found
33 |
34 | Returns:
35 | The value of the environment variable or the default
36 | """
37 | return os.environ.get(key, default)
38 |
39 | def get_api_key() -> str:
40 | """
41 | Get the fal.ai API key from environment variables.
42 |
43 | Returns:
44 | The API key as a string
45 |
46 | Raises:
47 | ValueError: If the FAL_KEY environment variable is not set
48 | """
49 | api_key = get_env(API_KEY_ENV_VAR)
50 | if not api_key:
51 | raise ValueError(f"{API_KEY_ENV_VAR} environment variable not set")
52 | return api_key
```
--------------------------------------------------------------------------------
/api/models.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Models module for fal.ai MCP server.
3 |
4 | This module provides tools for listing, searching,
5 | and retrieving schemas for fal.ai models.
6 | """
7 |
8 | from typing import Optional, Dict, Any, List
9 | from fastmcp import FastMCP
10 | from .utils import public_request
11 | from .config import FAL_BASE_URL
12 |
13 | def register_model_tools(mcp: FastMCP):
14 | """Register model-related tools with the MCP server."""
15 |
16 | @mcp.tool()
17 | async def models(page: Optional[int] = None, total: Optional[int] = None) -> List[Dict[str, Any]]:
18 | """
19 | List available models on fal.ai. Ensure to use the total and page arguments. Avoid listing all the models at once.
20 |
21 | Args:
22 | page: The page number of models to retrieve (pagination)
23 | total: The total number of models to retrieve per page
24 |
25 | Returns:
26 | A list of models with their metadata
27 | """
28 | url = f"{FAL_BASE_URL}/models"
29 |
30 | params = {}
31 | if page is not None:
32 | params["page"] = page
33 | if total is not None:
34 | params["total"] = total
35 |
36 | if params:
37 | url += "?" + "&".join(f"{k}={v}" for k, v in params.items())
38 |
39 | result = await public_request(url)
40 |
41 | return result
42 |
43 | @mcp.tool()
44 | async def search(keywords: str) -> List[Dict[str, Any]]:
45 | """
46 | Search for models on fal.ai based on keywords.
47 |
48 | Args:
49 | keywords: The search terms to find models
50 |
51 | Returns:
52 | A list of models matching the search criteria
53 | """
54 | if not isinstance(keywords, str):
55 | keywords = str(keywords)
56 |
57 | url = f"{FAL_BASE_URL}/models?keywords={keywords}"
58 |
59 | result = await public_request(url)
60 |
61 | return result
62 |
63 | @mcp.tool()
64 | async def schema(model_id: str) -> Dict[str, Any]:
65 | """
66 | Get the OpenAPI schema for a specific model.
67 |
68 | Args:
69 | model_id: The ID of the model (e.g., "fal-ai/flux/dev")
70 |
71 | Returns:
72 | The OpenAPI schema for the model
73 | """
74 | if not isinstance(model_id, str):
75 | model_id = str(model_id)
76 |
77 | url = f"{FAL_BASE_URL}/openapi/queue/openapi.json?endpoint_id={model_id}"
78 |
79 | return await public_request(url)
```
--------------------------------------------------------------------------------
/api/storage.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Storage module for fal.ai MCP server.
3 |
4 | This module provides tools for uploading files to fal.ai storage.
5 | """
6 |
7 | import os
8 | import mimetypes
9 | import aiofiles
10 | from typing import Dict, Any, Optional
11 | from fastmcp import FastMCP
12 | from .utils import authenticated_request, FalAPIError
13 | from .config import FAL_REST_URL
14 |
15 | def register_storage_tools(mcp: FastMCP):
16 | """Register storage-related tools with the MCP server."""
17 |
18 | @mcp.tool()
19 | async def upload(path: str) -> Dict[str, Any]:
20 | """
21 | Upload a file to fal.ai storage.
22 |
23 | Args:
24 | path: The absolute path to the file to upload
25 |
26 | Returns:
27 | Information about the uploaded file, including the file_url
28 | """
29 | if not os.path.exists(path):
30 | raise FileNotFoundError(f"File not found: {path}")
31 |
32 | filename = os.path.basename(path)
33 | file_size = os.path.getsize(path)
34 |
35 | content_type = mimetypes.guess_type(path)[0]
36 | if not content_type:
37 | content_type = "application/octet-stream"
38 |
39 | initiate_url = f"{FAL_REST_URL}/storage/upload/initiate?storage_type=fal-cdn-v3"
40 | initiate_payload = {
41 | "content_type": content_type,
42 | "file_name": filename
43 | }
44 |
45 | try:
46 | initiate_response = await authenticated_request(
47 | url=initiate_url,
48 | method="POST",
49 | json_data=initiate_payload
50 | )
51 |
52 | file_url = initiate_response["file_url"]
53 | upload_url = initiate_response["upload_url"]
54 |
55 | async with aiofiles.open(path, "rb") as file:
56 | file_content = await file.read()
57 |
58 | import httpx
59 | async with httpx.AsyncClient() as client:
60 | upload_response = await client.put(
61 | upload_url,
62 | content=file_content,
63 | headers={"Content-Type": content_type}
64 | )
65 | upload_response.raise_for_status()
66 |
67 | return {
68 | "file_url": file_url,
69 | "file_name": filename,
70 | "file_size": file_size,
71 | "content_type": content_type
72 | }
73 |
74 | except FalAPIError as e:
75 | raise
```
--------------------------------------------------------------------------------
/api/generate.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Generate module for fal.ai MCP server.
3 |
4 | This module provides tools for generating content
5 | and managing queue operations with fal.ai models.
6 | """
7 |
8 | from typing import Dict, Any, Optional
9 | from fastmcp import FastMCP
10 | from .utils import authenticated_request, sanitize_parameters, FalAPIError
11 | from .config import FAL_QUEUE_URL, FAL_DIRECT_URL
12 |
13 | def register_generation_tools(mcp: FastMCP):
14 | """Register generation-related tools with the MCP server."""
15 |
16 | @mcp.tool()
17 | async def generate(model: str, parameters: Dict[str, Any], queue: bool = False) -> Dict[str, Any]:
18 | """
19 | Generate content using a fal.ai model.
20 |
21 | Args:
22 | model: The model ID to use (e.g., "fal-ai/flux/dev")
23 | parameters: Model-specific parameters as a dictionary
24 | queue: Whether to use the queuing system (default: False)
25 |
26 | Returns:
27 | The model's response
28 | """
29 | if not isinstance(model, str):
30 | model = str(model)
31 |
32 | sanitized_parameters = sanitize_parameters(parameters)
33 |
34 | try:
35 | if queue:
36 | url = f"{FAL_QUEUE_URL}/{model}"
37 | else:
38 | url = f"{FAL_DIRECT_URL}/{model}"
39 |
40 | result = await authenticated_request(url, method="POST", json_data=sanitized_parameters)
41 |
42 | return result
43 |
44 | except FalAPIError as e:
45 | raise
46 |
47 | @mcp.tool()
48 | async def result(url: str) -> Dict[str, Any]:
49 | """
50 | Get the result of a queued request.
51 |
52 | Args:
53 | url: The response_url from a queued request
54 |
55 | Returns:
56 | The generation result
57 | """
58 | if not isinstance(url, str):
59 | url = str(url)
60 |
61 | try:
62 | result = await authenticated_request(url)
63 |
64 | return result
65 |
66 | except FalAPIError as e:
67 | raise
68 |
69 | @mcp.tool()
70 | async def status(url: str) -> Dict[str, Any]:
71 | """
72 | Check the status of a queued request.
73 |
74 | Args:
75 | url: The status_url from a queued request
76 |
77 | Returns:
78 | The current status of the queued request
79 | """
80 | if not isinstance(url, str):
81 | url = str(url)
82 |
83 | try:
84 | result = await authenticated_request(url)
85 |
86 | return result
87 |
88 | except FalAPIError as e:
89 | raise
90 |
91 | @mcp.tool()
92 | async def cancel(url: str) -> Dict[str, Any]:
93 | """
94 | Cancel a queued request.
95 |
96 | Args:
97 | url: The cancel_url from a queued request
98 |
99 | Returns:
100 | The result of the cancellation attempt
101 | """
102 | if not isinstance(url, str):
103 | url = str(url)
104 |
105 | try:
106 | result = await authenticated_request(url, method="PUT")
107 |
108 | return result
109 |
110 | except FalAPIError as e:
111 | raise
```
--------------------------------------------------------------------------------
/api/utils.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Utility functions for the fal.ai MCP server.
3 |
4 | This module provides helper functions for making API requests
5 | and handling authentication with the fal.ai service.
6 | """
7 |
8 | import httpx
9 | import json
10 | from typing import Optional, Dict, Any, Union
11 | from .config import get_api_key, DEFAULT_TIMEOUT, AUTHENTICATED_TIMEOUT
12 |
13 | class FalAPIError(Exception):
14 | """Exception raised for errors in the fal.ai API responses."""
15 | def __init__(self, message: str, status_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None):
16 | """
17 | Initialize the exception with error details.
18 |
19 | Args:
20 | message: The error message
21 | status_code: Optional HTTP status code
22 | details: Optional error details as a dictionary
23 | """
24 | self.status_code = status_code
25 | self.details = details or {}
26 | error_msg = message
27 | if status_code:
28 | error_msg = f"[{status_code}] {error_msg}"
29 | super().__init__(error_msg)
30 |
31 | async def authenticated_request(
32 | url: str,
33 | method: str = "GET",
34 | json_data: Optional[Dict[str, Any]] = None,
35 | timeout: float = AUTHENTICATED_TIMEOUT
36 | ) -> Dict[str, Any]:
37 | """
38 | Make an authenticated request to fal.ai API.
39 |
40 | Args:
41 | url: The URL to make the request to
42 | method: The HTTP method to use (GET, POST, PUT)
43 | json_data: Optional JSON data to send with the request
44 | timeout: Request timeout in seconds
45 |
46 | Returns:
47 | The JSON response from the API
48 |
49 | Raises:
50 | FalAPIError: If the API returns an error
51 | httpx.HTTPStatusError: If the HTTP request fails
52 | """
53 | try:
54 | headers = {"Authorization": f"Key {get_api_key()}"}
55 |
56 | async with httpx.AsyncClient() as client:
57 | if method == "GET":
58 | response = await client.get(url, headers=headers, timeout=timeout)
59 | elif method == "POST":
60 | response = await client.post(url, headers=headers, json=json_data, timeout=timeout)
61 | elif method == "PUT":
62 | response = await client.put(url, headers=headers, json=json_data, timeout=timeout)
63 | else:
64 | raise ValueError(f"Unsupported HTTP method: {method}")
65 |
66 | try:
67 | response.raise_for_status()
68 | except httpx.HTTPStatusError as e:
69 | try:
70 | error_details = json.loads(e.response.text)
71 | raise FalAPIError(
72 | f"API error: {error_details}",
73 | status_code=e.response.status_code,
74 | details=error_details
75 | )
76 | except json.JSONDecodeError:
77 | raise FalAPIError(
78 | f"API error: {e.response.text}",
79 | status_code=e.response.status_code
80 | )
81 |
82 | return response.json()
83 |
84 | except httpx.RequestError as e:
85 | raise FalAPIError(f"Request failed: {str(e)}")
86 |
87 | async def public_request(url: str, timeout: float = DEFAULT_TIMEOUT) -> Dict[str, Any]:
88 | """
89 | Make a non-authenticated request to fal.ai API.
90 |
91 | Args:
92 | url: The URL to make the request to
93 | timeout: Request timeout in seconds
94 |
95 | Returns:
96 | The JSON response from the API
97 |
98 | Raises:
99 | FalAPIError: If the API returns an error
100 | httpx.HTTPStatusError: If the HTTP request fails
101 | """
102 | try:
103 | async with httpx.AsyncClient() as client:
104 | response = await client.get(url, timeout=timeout)
105 |
106 | try:
107 | response.raise_for_status()
108 | except httpx.HTTPStatusError as e:
109 | try:
110 | error_details = json.loads(e.response.text)
111 | raise FalAPIError(
112 | f"API error: {error_details}",
113 | status_code=e.response.status_code,
114 | details=error_details
115 | )
116 | except json.JSONDecodeError:
117 | raise FalAPIError(
118 | f"API error: {e.response.text}",
119 | status_code=e.response.status_code
120 | )
121 |
122 | return response.json()
123 |
124 | except httpx.RequestError as e:
125 | raise FalAPIError(f"Request failed: {str(e)}")
126 |
127 | def sanitize_parameters(parameters: Dict[str, Any]) -> Dict[str, Any]:
128 | """
129 | Sanitize parameters for API requests.
130 |
131 | Args:
132 | parameters: The parameters to sanitize
133 |
134 | Returns:
135 | Sanitized parameters
136 | """
137 | sanitized = parameters.copy()
138 | sanitized = {k: v for k, v in sanitized.items() if v is not None}
139 |
140 | return sanitized
```