# Directory Structure
```
├── api
│ ├── __init__.py
│ ├── config.py
│ ├── generate.py
│ ├── models.py
│ ├── storage.py
│ └── utils.py
├── LICENSE
├── main.py
└── README.md
```
# Files
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# fal.ai MCP Server
A Model Context Protocol (MCP) server for interacting with fal.ai models and services.
## Features
- List all available fal.ai models
- Search for specific models by keywords
- Get model schemas
- Generate content using any fal.ai model
- Support for both direct and queued model execution
- Queue management (status checking, getting results, cancelling requests)
- File upload to fal.ai CDN
## Requirements
- Python 3.10+
- fastmcp
- httpx
- aiofiles
- A fal.ai API key
## Installation
1. Clone this repository:
```bash
git clone https://github.com/am0y/mcp-fal.git
cd mcp-fal
```
2. Install the required packages:
```bash
pip install fastmcp httpx aiofiles
```
3. Set your fal.ai API key as an environment variable:
```bash
export FAL_KEY="YOUR_FAL_API_KEY_HERE"
```
## Usage
### Running the Server
You can run the server in development mode with:
```bash
fastmcp dev main.py
```
This will launch the MCP Inspector web interface where you can test the tools interactively.
### Installing in Claude Desktop
To use the server with Claude Desktop:
```bash
fastmcp install main.py -e FAL_KEY="YOUR_FAL_API_KEY_HERE"
```
This will make the server available to Claude in the Desktop app.
### Running Directly
You can also run the server directly:
```bash
python main.py
```
## API Reference
### Tools
- `models(page=None, total=None)` - List available models with optional pagination
- `search(keywords)` - Search for models by keywords
- `schema(model_id)` - Get OpenAPI schema for a specific model
- `generate(model, parameters, queue=False)` - Generate content using a model
- `result(url)` - Get result from a queued request
- `status(url)` - Check status of a queued request
- `cancel(url)` - Cancel a queued request
- `upload(path` - Upload a file to fal.ai CDN
## License
[MIT](LICENSE)
```
--------------------------------------------------------------------------------
/api/__init__.py:
--------------------------------------------------------------------------------
```python
"""
fal.ai MCP Server API package.
This package contains modules for interacting with fal.ai models and services through the Model Context Protocol (MCP).
"""
__version__ = "1.0.0"
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
"""
fal.ai MCP Server : Main entry point
This module sets up and runs the fal.ai MCP server,
providing tools to interact with fal.ai models and services.
"""
import os
import sys
from fastmcp import FastMCP
from api.models import register_model_tools
from api.generate import register_generation_tools
from api.storage import register_storage_tools
from api.config import get_api_key, SERVER_NAME, SERVER_DESCRIPTION, SERVER_VERSION, SERVER_DEPENDENCIES
mcp = FastMCP(
SERVER_NAME,
description=SERVER_DESCRIPTION,
dependencies=SERVER_DEPENDENCIES,
version=SERVER_VERSION
)
register_model_tools(mcp)
register_generation_tools(mcp)
register_storage_tools(mcp)
def main():
try:
get_api_key()
except ValueError:
pass
try:
mcp.run()
except Exception as e:
sys.exit(1)
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/api/config.py:
--------------------------------------------------------------------------------
```python
"""
Configuration module for the fal.ai MCP server.
This module provides centralized configuration settings for the fal.ai MCP server, including API endpoints, timeouts, and more.
"""
import os
from typing import Dict, Any, Optional
FAL_BASE_URL = "https://fal.ai/api"
FAL_QUEUE_URL = "https://queue.fal.run"
FAL_DIRECT_URL = "https://fal.run"
FAL_REST_URL = "https://rest.alpha.fal.ai"
DEFAULT_TIMEOUT = 30.0
AUTHENTICATED_TIMEOUT = 100.0
API_KEY_ENV_VAR = "FAL_KEY"
SERVER_NAME = "fal.ai MCP Server"
SERVER_DESCRIPTION = "Access fal.ai models and generate content through MCP"
SERVER_VERSION = "1.0.0"
SERVER_DEPENDENCIES = ["httpx", "aiofiles"]
def get_env(key: str, default: Optional[str] = None) -> Optional[str]:
"""
Get an environment variable with optional default value.
Args:
key: The name of the environment variable
default: Optional default value if not found
Returns:
The value of the environment variable or the default
"""
return os.environ.get(key, default)
def get_api_key() -> str:
"""
Get the fal.ai API key from environment variables.
Returns:
The API key as a string
Raises:
ValueError: If the FAL_KEY environment variable is not set
"""
api_key = get_env(API_KEY_ENV_VAR)
if not api_key:
raise ValueError(f"{API_KEY_ENV_VAR} environment variable not set")
return api_key
```
--------------------------------------------------------------------------------
/api/models.py:
--------------------------------------------------------------------------------
```python
"""
Models module for fal.ai MCP server.
This module provides tools for listing, searching,
and retrieving schemas for fal.ai models.
"""
from typing import Optional, Dict, Any, List
from fastmcp import FastMCP
from .utils import public_request
from .config import FAL_BASE_URL
def register_model_tools(mcp: FastMCP):
"""Register model-related tools with the MCP server."""
@mcp.tool()
async def models(page: Optional[int] = None, total: Optional[int] = None) -> List[Dict[str, Any]]:
"""
List available models on fal.ai. Ensure to use the total and page arguments. Avoid listing all the models at once.
Args:
page: The page number of models to retrieve (pagination)
total: The total number of models to retrieve per page
Returns:
A list of models with their metadata
"""
url = f"{FAL_BASE_URL}/models"
params = {}
if page is not None:
params["page"] = page
if total is not None:
params["total"] = total
if params:
url += "?" + "&".join(f"{k}={v}" for k, v in params.items())
result = await public_request(url)
return result
@mcp.tool()
async def search(keywords: str) -> List[Dict[str, Any]]:
"""
Search for models on fal.ai based on keywords.
Args:
keywords: The search terms to find models
Returns:
A list of models matching the search criteria
"""
if not isinstance(keywords, str):
keywords = str(keywords)
url = f"{FAL_BASE_URL}/models?keywords={keywords}"
result = await public_request(url)
return result
@mcp.tool()
async def schema(model_id: str) -> Dict[str, Any]:
"""
Get the OpenAPI schema for a specific model.
Args:
model_id: The ID of the model (e.g., "fal-ai/flux/dev")
Returns:
The OpenAPI schema for the model
"""
if not isinstance(model_id, str):
model_id = str(model_id)
url = f"{FAL_BASE_URL}/openapi/queue/openapi.json?endpoint_id={model_id}"
return await public_request(url)
```
--------------------------------------------------------------------------------
/api/storage.py:
--------------------------------------------------------------------------------
```python
"""
Storage module for fal.ai MCP server.
This module provides tools for uploading files to fal.ai storage.
"""
import os
import mimetypes
import aiofiles
from typing import Dict, Any, Optional
from fastmcp import FastMCP
from .utils import authenticated_request, FalAPIError
from .config import FAL_REST_URL
def register_storage_tools(mcp: FastMCP):
"""Register storage-related tools with the MCP server."""
@mcp.tool()
async def upload(path: str) -> Dict[str, Any]:
"""
Upload a file to fal.ai storage.
Args:
path: The absolute path to the file to upload
Returns:
Information about the uploaded file, including the file_url
"""
if not os.path.exists(path):
raise FileNotFoundError(f"File not found: {path}")
filename = os.path.basename(path)
file_size = os.path.getsize(path)
content_type = mimetypes.guess_type(path)[0]
if not content_type:
content_type = "application/octet-stream"
initiate_url = f"{FAL_REST_URL}/storage/upload/initiate?storage_type=fal-cdn-v3"
initiate_payload = {
"content_type": content_type,
"file_name": filename
}
try:
initiate_response = await authenticated_request(
url=initiate_url,
method="POST",
json_data=initiate_payload
)
file_url = initiate_response["file_url"]
upload_url = initiate_response["upload_url"]
async with aiofiles.open(path, "rb") as file:
file_content = await file.read()
import httpx
async with httpx.AsyncClient() as client:
upload_response = await client.put(
upload_url,
content=file_content,
headers={"Content-Type": content_type}
)
upload_response.raise_for_status()
return {
"file_url": file_url,
"file_name": filename,
"file_size": file_size,
"content_type": content_type
}
except FalAPIError as e:
raise
```
--------------------------------------------------------------------------------
/api/generate.py:
--------------------------------------------------------------------------------
```python
"""
Generate module for fal.ai MCP server.
This module provides tools for generating content
and managing queue operations with fal.ai models.
"""
from typing import Dict, Any, Optional
from fastmcp import FastMCP
from .utils import authenticated_request, sanitize_parameters, FalAPIError
from .config import FAL_QUEUE_URL, FAL_DIRECT_URL
def register_generation_tools(mcp: FastMCP):
"""Register generation-related tools with the MCP server."""
@mcp.tool()
async def generate(model: str, parameters: Dict[str, Any], queue: bool = False) -> Dict[str, Any]:
"""
Generate content using a fal.ai model.
Args:
model: The model ID to use (e.g., "fal-ai/flux/dev")
parameters: Model-specific parameters as a dictionary
queue: Whether to use the queuing system (default: False)
Returns:
The model's response
"""
if not isinstance(model, str):
model = str(model)
sanitized_parameters = sanitize_parameters(parameters)
try:
if queue:
url = f"{FAL_QUEUE_URL}/{model}"
else:
url = f"{FAL_DIRECT_URL}/{model}"
result = await authenticated_request(url, method="POST", json_data=sanitized_parameters)
return result
except FalAPIError as e:
raise
@mcp.tool()
async def result(url: str) -> Dict[str, Any]:
"""
Get the result of a queued request.
Args:
url: The response_url from a queued request
Returns:
The generation result
"""
if not isinstance(url, str):
url = str(url)
try:
result = await authenticated_request(url)
return result
except FalAPIError as e:
raise
@mcp.tool()
async def status(url: str) -> Dict[str, Any]:
"""
Check the status of a queued request.
Args:
url: The status_url from a queued request
Returns:
The current status of the queued request
"""
if not isinstance(url, str):
url = str(url)
try:
result = await authenticated_request(url)
return result
except FalAPIError as e:
raise
@mcp.tool()
async def cancel(url: str) -> Dict[str, Any]:
"""
Cancel a queued request.
Args:
url: The cancel_url from a queued request
Returns:
The result of the cancellation attempt
"""
if not isinstance(url, str):
url = str(url)
try:
result = await authenticated_request(url, method="PUT")
return result
except FalAPIError as e:
raise
```
--------------------------------------------------------------------------------
/api/utils.py:
--------------------------------------------------------------------------------
```python
"""
Utility functions for the fal.ai MCP server.
This module provides helper functions for making API requests
and handling authentication with the fal.ai service.
"""
import httpx
import json
from typing import Optional, Dict, Any, Union
from .config import get_api_key, DEFAULT_TIMEOUT, AUTHENTICATED_TIMEOUT
class FalAPIError(Exception):
"""Exception raised for errors in the fal.ai API responses."""
def __init__(self, message: str, status_code: Optional[int] = None, details: Optional[Dict[str, Any]] = None):
"""
Initialize the exception with error details.
Args:
message: The error message
status_code: Optional HTTP status code
details: Optional error details as a dictionary
"""
self.status_code = status_code
self.details = details or {}
error_msg = message
if status_code:
error_msg = f"[{status_code}] {error_msg}"
super().__init__(error_msg)
async def authenticated_request(
url: str,
method: str = "GET",
json_data: Optional[Dict[str, Any]] = None,
timeout: float = AUTHENTICATED_TIMEOUT
) -> Dict[str, Any]:
"""
Make an authenticated request to fal.ai API.
Args:
url: The URL to make the request to
method: The HTTP method to use (GET, POST, PUT)
json_data: Optional JSON data to send with the request
timeout: Request timeout in seconds
Returns:
The JSON response from the API
Raises:
FalAPIError: If the API returns an error
httpx.HTTPStatusError: If the HTTP request fails
"""
try:
headers = {"Authorization": f"Key {get_api_key()}"}
async with httpx.AsyncClient() as client:
if method == "GET":
response = await client.get(url, headers=headers, timeout=timeout)
elif method == "POST":
response = await client.post(url, headers=headers, json=json_data, timeout=timeout)
elif method == "PUT":
response = await client.put(url, headers=headers, json=json_data, timeout=timeout)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
try:
response.raise_for_status()
except httpx.HTTPStatusError as e:
try:
error_details = json.loads(e.response.text)
raise FalAPIError(
f"API error: {error_details}",
status_code=e.response.status_code,
details=error_details
)
except json.JSONDecodeError:
raise FalAPIError(
f"API error: {e.response.text}",
status_code=e.response.status_code
)
return response.json()
except httpx.RequestError as e:
raise FalAPIError(f"Request failed: {str(e)}")
async def public_request(url: str, timeout: float = DEFAULT_TIMEOUT) -> Dict[str, Any]:
"""
Make a non-authenticated request to fal.ai API.
Args:
url: The URL to make the request to
timeout: Request timeout in seconds
Returns:
The JSON response from the API
Raises:
FalAPIError: If the API returns an error
httpx.HTTPStatusError: If the HTTP request fails
"""
try:
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=timeout)
try:
response.raise_for_status()
except httpx.HTTPStatusError as e:
try:
error_details = json.loads(e.response.text)
raise FalAPIError(
f"API error: {error_details}",
status_code=e.response.status_code,
details=error_details
)
except json.JSONDecodeError:
raise FalAPIError(
f"API error: {e.response.text}",
status_code=e.response.status_code
)
return response.json()
except httpx.RequestError as e:
raise FalAPIError(f"Request failed: {str(e)}")
def sanitize_parameters(parameters: Dict[str, Any]) -> Dict[str, Any]:
"""
Sanitize parameters for API requests.
Args:
parameters: The parameters to sanitize
Returns:
Sanitized parameters
"""
sanitized = parameters.copy()
sanitized = {k: v for k, v in sanitized.items() if v is not None}
return sanitized
```