# Directory Structure
```
├── .env.example
├── .github
│ ├── funding.yml
│ └── workflows
│ └── docker-build.yml
├── .gitignore
├── build-all.sh
├── build-http.sh
├── build-sse.sh
├── build-stdio.sh
├── build.sh
├── docker-compose.yml
├── Dockerfile.http
├── Dockerfile.sse
├── Dockerfile.stdio
├── examples
│ └── basic_usage.py
├── LICENSE
├── main_http.py
├── main.py
├── pyproject.toml
├── README.md
├── src
│ └── novareel_mcp_server
│ ├── __init__.py
│ ├── prompting_guide.py
│ ├── server_http.py
│ ├── server_sse.py
│ └── server.py
└── start.sh
```
# Files
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# Amazon Nova Reel MCP Server Configuration
# Copy this file to .env and fill in your actual values
# AWS Credentials (Option 1: Explicit credentials)
AWS_ACCESS_KEY_ID=your_aws_access_key_id_here
AWS_SECRET_ACCESS_KEY=your_aws_secret_access_key_here
AWS_SESSION_TOKEN=your_session_token_here_if_using_temporary_credentials
# AWS Credentials (Option 2: Use AWS Profile)
# AWS_PROFILE=your_aws_profile_name
# AWS Configuration
AWS_REGION=us-east-1
# S3 Bucket for video output (Required)
# This bucket must exist and be accessible with your AWS credentials
S3_BUCKET=your-video-generation-bucket-name
# Example values:
# AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE
# AWS_SECRET_ACCESS_KEY=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
# AWS_SESSION_TOKEN=IQoJb3JpZ2luX2VjEHoaCXVzLWVhc3QtMSJHMEUCIQD...
# AWS_PROFILE=my-profile
# AWS_REGION=us-east-1
# S3_BUCKET=my-nova-reel-videos
# Notes:
# - Never commit the actual .env file with real credentials to version control
# - You can use either explicit credentials OR an AWS profile, not both
# - For temporary credentials (STS), include AWS_SESSION_TOKEN
# - For local development, AWS_PROFILE is often more convenient
# - Ensure your AWS credentials have the necessary permissions for Bedrock and S3
# - The S3 bucket should be in the same region as your Bedrock service
# - Videos will be stored in the bucket with structure: s3://bucket/job-id/output.mp4
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
pip-wheel-metadata/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# AWS credentials (never commit these!)
.aws/
aws-credentials.json
credentials.json
# Docker
.dockerignore
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# OS
.DS_Store
.DS_Store?
._*
.Spotlight-V100
.Trashes
ehthumbs.db
Thumbs.db
# Logs
*.log
logs/
# Temporary files
tmp/
temp/
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Amazon Nova Reel 1.1 MCP Server
A Model Context Protocol (MCP) server for Amazon Nova Reel 1.1 video generation using AWS Bedrock. This server provides tools for asynchronous video generation with comprehensive prompting guidelines and both stdio and SSE transport support.
<div align="center">
[](https://www.buymeacoffee.com/mirecekdg)
</div>
## Features
- **Asynchronous Video Generation**: Start, monitor, and retrieve video generation jobs
- **Multiple Transport Methods**: Support for stdio, Server-Sent Events (SSE), and HTTP Streaming
- **Comprehensive Prompting Guide**: Built-in guidelines based on AWS documentation
- **Docker Support**: Ready-to-use Docker containers for all transport methods
- **AWS Integration**: Full integration with AWS Bedrock and S3
## Available Tools
### 1. `start_async_invoke`
Start a new video generation job.
**Parameters:**
- `prompt` (required): Text description for video generation
- `duration_seconds` (optional): Video duration (12-120 seconds, multiples of 6, default: 12)
- `fps` (optional): Frames per second (default: 24)
- `dimension` (optional): Video dimensions (default: "1280x720")
- `seed` (optional): Random seed for reproducible results
- `task_type` (optional): Task type (default: "MULTI_SHOT_AUTOMATED")
**Returns:** Job details including `job_id`, `invocation_arn`, and estimated video URL.
### 2. `list_async_invokes`
List all tracked video generation jobs with their current status.
**Returns:** Summary of all jobs with status counts and individual job details.
### 3. `get_async_invoke`
Get detailed information about a specific video generation job.
**Parameters:**
- `identifier` (required): Either `job_id` or `invocation_arn`
**Returns:** Detailed job information including video URL when completed.
### 4. `get_prompting_guide`
Get comprehensive prompting guidelines for effective video generation.
**Returns:** Detailed prompting best practices, examples, and templates.
## Installation
### Prerequisites
- Python 3.8+
- AWS Account with Bedrock access
- S3 bucket for video output
- AWS credentials with appropriate permissions
### Local Installation
1. Clone or download the server files
2. Install dependencies:
```bash
pip install -e .
```
### Docker Installation
#### Using Pre-built Images (Recommended)
Pull multi-architecture images from GitHub Container Registry:
```bash
# STDIO version
docker pull ghcr.io/mirecekd/novareel-mcp:latest-stdio
# SSE version
docker pull ghcr.io/mirecekd/novareel-mcp:latest-sse
# HTTP Streaming version
docker pull ghcr.io/mirecekd/novareel-mcp:latest-http
```
#### Building Locally
1. Build containers using provided scripts:
```bash
# Build all versions
./build-all.sh
# Or build individual versions
./build-stdio.sh # STDIO version
./build-sse.sh # SSE version
./build-http.sh # HTTP Streaming version
```
2. Or use docker-compose:
```bash
docker-compose up -d
```
3. Or use the quick start script:
```bash
# Build all images
./start.sh build
# Build specific version
./start.sh build-stdio
./start.sh build-sse
./start.sh build-http
```
## Configuration
### Environment Variables
- `AWS_ACCESS_KEY_ID`: Your AWS access key ID
- `AWS_SECRET_ACCESS_KEY`: Your AWS secret access key
- `AWS_REGION`: AWS region (default: us-east-1)
- `S3_BUCKET`: S3 bucket name for video output
### .env File Example
Create a `.env` file for docker-compose:
```env
AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE
AWS_SECRET_ACCESS_KEY=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
AWS_REGION=us-east-1
S3_BUCKET=my-video-generation-bucket
```
## Usage
### MCP Client Integration (Cline/Claude Desktop)
Add the server to your MCP client configuration:
#### Cline Configuration
Add to your Cline MCP settings:
```json
{
"mcpServers": {
"Nova Reel Video MCP": {
"disabled": false,
"timeout": 60,
"type": "stdio",
"command": "docker",
"args": [
"run",
"-i",
"--rm",
"ghcr.io/mirecekd/novareel-mcp:latest-stdio",
"--aws-access-key-id",
"YOUR_AWS_ACCESS_KEY_ID",
"--aws-secret-access-key",
"YOUR_AWS_SECRET_ACCESS_KEY",
"--s3-bucket",
"YOUR_S3_BUCKET_NAME"
]
}
}
}
```
#### Claude Desktop Configuration
Add to your Claude Desktop `claude_desktop_config.json`:
```json
{
"mcpServers": {
"novareel-mcp": {
"command": "docker",
"args": [
"run",
"-i",
"--rm",
"ghcr.io/mirecekd/novareel-mcp:latest-stdio",
"--aws-access-key-id",
"YOUR_AWS_ACCESS_KEY_ID",
"--aws-secret-access-key",
"YOUR_AWS_SECRET_ACCESS_KEY",
"--s3-bucket",
"YOUR_S3_BUCKET_NAME"
]
}
}
}
```
#### Alternative: Local Python Installation
If you prefer running without Docker:
```json
{
"mcpServers": {
"novareel-mcp": {
"command": "uvx",
"args": [
"--from", "git+https://github.com/mirecekd/novareel-mcp.git",
"novareel-mcp-server",
"--aws-access-key-id", "YOUR_AWS_ACCESS_KEY_ID",
"--aws-secret-access-key", "YOUR_AWS_SECRET_ACCESS_KEY",
"--s3-bucket", "YOUR_S3_BUCKET_NAME"
]
}
}
}
```
**Important**: Replace the placeholder values with your actual AWS credentials and S3 bucket name.
### Running with uvx (Recommended)
```bash
# First build the package
./build.sh
# Then run from wheel file
uvx --from ./dist/novareel_mcp-1.0.0-py3-none-any.whl novareel-mcp-server --aws-access-key-id YOUR_KEY --aws-secret-access-key YOUR_SECRET --s3-bucket YOUR_BUCKET
# Or from current directory during development (without build)
uvx --from . novareel-mcp-server --aws-access-key-id YOUR_KEY --aws-secret-access-key YOUR_SECRET --s3-bucket YOUR_BUCKET
# Or using start script
./start.sh build-package # Build wheel
```
### Stdio Version (Direct MCP Client)
```bash
# Local execution
python main.py --aws-access-key-id YOUR_KEY --aws-secret-access-key YOUR_SECRET --s3-bucket YOUR_BUCKET
# Docker execution
docker run --rm -i mirecekd/novareel-mcp-server:stdio --aws-access-key-id YOUR_KEY --aws-secret-access-key YOUR_SECRET --s3-bucket YOUR_BUCKET
```
### SSE Version (Web Interface)
```bash
# Local execution
python -m novareel_mcp_server.server_sse --aws-access-key-id YOUR_KEY --aws-secret-access-key YOUR_SECRET --s3-bucket YOUR_BUCKET --host 0.0.0.0 --port 8000
# Docker execution
docker run -p 8000:8000 -e AWS_ACCESS_KEY_ID=YOUR_KEY -e AWS_SECRET_ACCESS_KEY=YOUR_SECRET -e S3_BUCKET=YOUR_BUCKET mirecekd/novareel-mcp-server:sse
```
Then access `http://localhost:8000/sse/` for the SSE endpoint.
### HTTP Streaming Version (Bidirectional Transport)
```bash
# Local execution
python -m novareel_mcp_server.server_http --aws-access-key-id YOUR_KEY --aws-secret-access-key YOUR_SECRET --s3-bucket YOUR_BUCKET --host 0.0.0.0 --port 8001
# Docker execution
docker run -p 8001:8001 -e AWS_ACCESS_KEY_ID=YOUR_KEY -e AWS_SECRET_ACCESS_KEY=YOUR_SECRET -e S3_BUCKET=YOUR_BUCKET ghcr.io/mirecekd/novareel-mcp:latest-http
```
Then access `http://localhost:8001` for the HTTP streaming transport.
### Package Build
To create a distribution package:
```bash
# Install build tools
pip install build
# Create package
python3 -m build
# Output files will be in dist/
```
## Example Usage
### Basic Video Generation
```python
# Start a video generation job
result = start_async_invoke(
prompt="A majestic eagle soars over a mountain valley, camera tracking its flight as it circles above a pristine lake",
duration_seconds=24,
fps=24,
dimension="1920x1080"
)
job_id = result["job_id"]
print(f"Started job: {job_id}")
# Check job status
status = get_async_invoke(job_id)
print(f"Status: {status['status']}")
# When completed, get video URL
if status["status"] == "Completed":
print(f"Video URL: {status['video_url']}")
```
### List All Jobs
```python
# Get overview of all jobs
jobs = list_async_invokes()
print(f"Total jobs: {jobs['total_invocations']}")
print(f"Completed: {jobs['summary']['completed']}")
print(f"In progress: {jobs['summary']['in_progress']}")
```
## Prompting Guidelines
The server includes comprehensive prompting guidelines based on AWS documentation. Access them using:
```python
guide = get_prompting_guide()
```
### Key Prompting Tips
1. **Be Specific**: Use detailed, descriptive language
- Good: "A red cardinal perched on a snow-covered pine branch, morning sunlight filtering through the trees"
- Bad: "A bird on a tree"
2. **Use Camera Terminology**: Control shot composition
- "Close-up shot of hands carving wood"
- "Wide shot establishing the mountain landscape"
- "Camera pans left across the valley"
3. **Include Lighting Details**: Specify atmosphere
- "Golden hour lighting casting long shadows"
- "Soft blue hour twilight"
- "Dramatic storm clouds overhead"
4. **Structure for Duration**: Match complexity to video length
- 12-24 seconds: Single action or moment
- 30-60 seconds: 2-3 distinct actions
- 60-120 seconds: Full narrative with multiple scenes
### Example Prompts by Category
**Nature (Short - 12s):**
```
Close-up of morning dew drops on a spider web, with soft sunrise lighting creating rainbow reflections
```
**Urban (Medium - 30s):**
```
A street musician plays violin in a subway station, commuters pause to listen, coins drop into his case, camera slowly pulls back to reveal the bustling underground scene
```
**Portrait (Long - 60s):**
```
Portrait of a chef preparing a signature dish: selecting fresh ingredients at market, returning to kitchen, methodically preparing each component, plating with artistic precision, and presenting the finished masterpiece
```
## AWS Permissions
Your AWS credentials need the following permissions:
```json
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel",
"bedrock:StartAsyncInvoke",
"bedrock:GetAsyncInvoke",
"bedrock:ListFoundationModels"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": [
"s3:PutObject",
"s3:GetObject",
"s3:ListBucket"
],
"Resource": [
"arn:aws:s3:::your-bucket-name",
"arn:aws:s3:::your-bucket-name/*"
]
}
]
}
```
## Video Output
Generated videos are stored in your S3 bucket with the following structure:
```
s3://your-bucket/
├── job-id-1/
│ └── output.mp4
├── job-id-2/
│ └── output.mp4
└── ...
```
Videos are accessible via HTTPS URLs:
```
https://your-bucket.s3.region.amazonaws.com/job-id/output.mp4
```
## Supported Video Specifications
- **Duration**: 12-120 seconds (must be multiples of 6)
- **Frame Rate**: 24 fps (recommended)
- **Dimensions**:
- 1280x720 (HD)
- **Format**: MP4
- **Model**: amazon.nova-reel-v1:1
## Troubleshooting
### Common Issues
1. **AWS Credentials Error**
- Verify your AWS credentials are correct
- Ensure your account has Bedrock access enabled
- Check IAM permissions
2. **S3 Bucket Access**
- Verify bucket exists and is accessible
- Check bucket permissions
- Ensure bucket is in the same region as Bedrock
3. **Duration Validation**
- Duration must be 12-120 seconds
- Must be a multiple of 6
- Valid values: 12, 18, 24, 30, 36, 42, 48, 54, 60, 66, 72, 78, 84, 90, 96, 102, 108, 114, 120
4. **Job Not Found**
- Use `list_async_invokes` to see all tracked jobs
- Jobs are stored in memory and lost on server restart
- For production, implement persistent storage
### Debug Mode
Enable debug logging by setting environment variable:
```bash
export PYTHONUNBUFFERED=1
```
## Development
### Project Structure
```
novareel-mcp-server/
├── main.py # Main MCP server (stdio)
├── main_sse.py # SSE version of MCP server
├── main_http.py # HTTP Streaming version of MCP server
├── prompting_guide.py # AWS prompting guidelines
├── pyproject.toml # Python dependencies
├── Dockerfile.stdio # Docker for stdio version
├── Dockerfile.sse # Docker for SSE version
├── Dockerfile.http # Docker for HTTP streaming version
├── docker-compose.yml # Container orchestration
└── README.md # This documentation
```
### Contributing
1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Test with all transport versions (stdio, SSE, HTTP streaming)
5. Submit a pull request
## License
This project is licensed under the MIT License - see the LICENSE file for details.
## Support
For issues and questions:
1. Check the troubleshooting section
2. Review AWS Bedrock documentation
3. Open an issue in the repository
## Related Links
- [AWS Nova Reel Documentation](https://docs.aws.amazon.com/nova/latest/userguide/)
- [Video Generation Prompting Guide](https://docs.aws.amazon.com/nova/latest/userguide/prompting-video-generation.html)
- [Camera Control Prompting](https://docs.aws.amazon.com/nova/latest/userguide/prompting-video-camera-control.html)
- [Model Context Protocol](https://modelcontextprotocol.io/)
- [FastMCP Framework](https://github.com/jlowin/fastmcp)
```
--------------------------------------------------------------------------------
/.github/funding.yml:
--------------------------------------------------------------------------------
```yaml
buy_me_a_coffee: mirecekdg
```
--------------------------------------------------------------------------------
/src/novareel_mcp_server/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Nova Reel MCP Server
MCP Server pro Amazon Nova Reel 1.1 video generation
"""
__version__ = "1.0.0"
```
--------------------------------------------------------------------------------
/main_http.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Amazon Nova Reel 1.1 MCP Server - HTTP Streaming Entry Point
"""
from src.novareel_mcp_server.server_http import main
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Nova Reel MCP Server - Main Entry Point
Wrapper pro zpětnou kompatibilitu
"""
from src.novareel_mcp_server.server import main
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/build.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
echo "Building Nova Reel MCP Server Python Package"
echo "============================================"
# Check if build module is installed
if ! python3 -c "import build" 2>/dev/null; then
echo "Installing build module..."
pip install build
fi
# Clean previous builds
echo "Cleaning previous builds..."
rm -rf dist/ build/ *.egg-info/
# Build the package
echo "Building package..."
python3 -m build
echo ""
echo "✅ Build completed!"
echo "Generated files:"
ls -la dist/
echo ""
echo "Usage with uvx:"
echo " uvx --from ./dist/novareel_mcp-1.0.0-py3-none-any.whl novareel-mcp-server --help"
echo ""
echo "Or install locally:"
echo " pip install ./dist/novareel_mcp-1.0.0-py3-none-any.whl"
```
--------------------------------------------------------------------------------
/build-stdio.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
echo "Building Nova Reel MCP Server - STDIO Version (stdio transport)"
echo "================================================================"
docker build -f Dockerfile.stdio -t mirecekd/novareel-mcp-server:stdio -t mirecekd/novareel-mcp-server:latest .
echo ""
echo "Build completed!"
echo "STDIO Version tags:"
echo " - mirecekd/novareel-mcp-server:stdio"
echo " - mirecekd/novareel-mcp-server:latest"
echo ""
echo "Usage examples:"
echo ""
echo " Explicit credentials:"
echo " docker run --rm -i mirecekd/novareel-mcp-server:stdio --aws-access-key-id YOUR_KEY --aws-secret-access-key YOUR_SECRET --s3-bucket YOUR_BUCKET"
echo ""
echo " With session token (temporary credentials):"
echo " docker run --rm -i mirecekd/novareel-mcp-server:stdio --aws-access-key-id YOUR_KEY --aws-secret-access-key YOUR_SECRET --aws-session-token YOUR_TOKEN --s3-bucket YOUR_BUCKET"
echo ""
echo " With AWS profile:"
echo " docker run --rm -i -v ~/.aws:/root/.aws mirecekd/novareel-mcp-server:stdio --aws-profile my-profile --s3-bucket YOUR_BUCKET"
```
--------------------------------------------------------------------------------
/build-sse.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
echo "Building Nova Reel MCP Server - SSE Version (HTTP transport)"
echo "============================================================"
docker build -f Dockerfile.sse -t mirecekd/novareel-mcp-server:sse -t mirecekd/novareel-mcp-sse .
echo ""
echo "Build completed!"
echo "SSE Version tags:"
echo " - mirecekd/novareel-mcp-server:sse"
echo " - mirecekd/novareel-mcp-sse"
echo ""
echo "Usage examples:"
echo ""
echo " Environment variables:"
echo " docker run -e AWS_ACCESS_KEY_ID=YOUR_KEY -e AWS_SECRET_ACCESS_KEY=YOUR_SECRET -e S3_BUCKET=YOUR_BUCKET -p 8000:8000 mirecekd/novareel-mcp-server:sse"
echo ""
echo " With session token (temporary credentials):"
echo " docker run -e AWS_ACCESS_KEY_ID=YOUR_KEY -e AWS_SECRET_ACCESS_KEY=YOUR_SECRET -e AWS_SESSION_TOKEN=YOUR_TOKEN -e S3_BUCKET=YOUR_BUCKET -p 8000:8000 mirecekd/novareel-mcp-server:sse"
echo ""
echo " With AWS profile:"
echo " docker run -v ~/.aws:/root/.aws -e AWS_PROFILE=my-profile -e S3_BUCKET=YOUR_BUCKET -p 8000:8000 mirecekd/novareel-mcp-server:sse"
echo ""
echo "SSE server will be available at: http://localhost:8000"
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "novareel-mcp"
version = "1.0.0"
description = "MCP Server pro Amazon Nova Reel 1.1 video generation"
authors = [
{name = "Miroslav Dvořák", email = "[email protected]"}
]
readme = "README.md"
license = {file = "LICENSE"}
requires-python = ">=3.8"
dependencies = [
"fastmcp>=0.2.0",
"boto3>=1.35.0",
"botocore>=1.35.0",
]
classifiers = [
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
]
[project.scripts]
novareel-mcp-server = "novareel_mcp_server.server:main"
[project.urls]
Homepage = "https://github.com/mirecekd/novareel-mcp"
Repository = "https://github.com/mirecekd/novareel-mcp"
Issues = "https://github.com/mirecekd/novareel-mcp/issues"
[tool.hatch.build.targets.wheel]
packages = ["src/novareel_mcp_server"]
[tool.hatch.build.targets.sdist]
include = [
"/src",
"/README.md",
"/LICENSE",
]
```
--------------------------------------------------------------------------------
/build-http.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# Build script for NovaReel MCP Server - HTTP Streaming version
set -e
echo "Building NovaReel MCP Server - HTTP Streaming version..."
# Build Docker image
docker build -f Dockerfile.http -t mirecekd/novareel-mcp-server:http .
echo "Build completed successfully!"
echo "Image: mirecekd/novareel-mcp-server:http"
echo ""
echo "Usage examples:"
echo ""
echo " Environment variables:"
echo " docker run -p 8001:8001 -e AWS_ACCESS_KEY_ID=your_key -e AWS_SECRET_ACCESS_KEY=your_secret -e S3_BUCKET=your_bucket mirecekd/novareel-mcp-server:http"
echo ""
echo " With session token (temporary credentials):"
echo " docker run -p 8001:8001 -e AWS_ACCESS_KEY_ID=your_key -e AWS_SECRET_ACCESS_KEY=your_secret -e AWS_SESSION_TOKEN=your_token -e S3_BUCKET=your_bucket mirecekd/novareel-mcp-server:http"
echo ""
echo " With AWS profile:"
echo " docker run -p 8001:8001 -v ~/.aws:/root/.aws -e AWS_PROFILE=my-profile -e S3_BUCKET=your_bucket mirecekd/novareel-mcp-server:http"
echo ""
echo " Command line arguments:"
echo " docker run -p 8001:8001 mirecekd/novareel-mcp-server:http --aws-access-key-id your_key --aws-secret-access-key your_secret --s3-bucket your_bucket"
echo ""
echo " Command line with session token:"
echo " docker run -p 8001:8001 mirecekd/novareel-mcp-server:http --aws-access-key-id your_key --aws-secret-access-key your_secret --aws-session-token your_token --s3-bucket your_bucket"
echo ""
echo " Command line with AWS profile:"
echo " docker run -p 8001:8001 -v ~/.aws:/root/.aws mirecekd/novareel-mcp-server:http --aws-profile my-profile --s3-bucket your_bucket"
echo ""
echo "HTTP streaming server will be available at: http://localhost:8001"
```
--------------------------------------------------------------------------------
/build-all.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
echo "Building All Nova Reel MCP Server Versions"
echo "=========================================="
echo ""
echo "1. Building STDIO Version (stdio transport)..."
echo "---------------------------------------------"
docker build -f Dockerfile.stdio -t mirecekd/novareel-mcp-server:stdio -t mirecekd/novareel-mcp-server:latest .
echo ""
echo "2. Building SSE Version (Server-Sent Events transport)..."
echo "--------------------------------------------------------"
docker build -f Dockerfile.sse -t mirecekd/novareel-mcp-server:sse -t mirecekd/novareel-mcp-sse .
echo ""
echo "3. Building HTTP Version (HTTP Streaming transport)..."
echo "-----------------------------------------------------"
docker build -f Dockerfile.http -t mirecekd/novareel-mcp-server:http .
echo ""
echo "✅ All builds completed!"
echo "========================"
echo ""
echo "Available images:"
echo " STDIO Version:"
echo " - mirecekd/novareel-mcp-server:stdio"
echo " - mirecekd/novareel-mcp-server:latest"
echo " SSE Version:"
echo " - mirecekd/novareel-mcp-server:sse"
echo " - mirecekd/novareel-mcp-sse"
echo " HTTP Streaming Version:"
echo " - mirecekd/novareel-mcp-server:http"
echo ""
echo "Usage examples:"
echo ""
echo " STDIO (Explicit credentials):"
echo " docker run --rm -i mirecekd/novareel-mcp-server:stdio --aws-access-key-id KEY --aws-secret-access-key SECRET --s3-bucket BUCKET"
echo ""
echo " STDIO (With session token):"
echo " docker run --rm -i mirecekd/novareel-mcp-server:stdio --aws-access-key-id KEY --aws-secret-access-key SECRET --aws-session-token TOKEN --s3-bucket BUCKET"
echo ""
echo " STDIO (With AWS profile):"
echo " docker run --rm -i -v ~/.aws:/root/.aws mirecekd/novareel-mcp-server:stdio --aws-profile my-profile --s3-bucket BUCKET"
echo ""
echo " SSE (Environment variables):"
echo " docker run -e AWS_ACCESS_KEY_ID=KEY -e AWS_SECRET_ACCESS_KEY=SECRET -e S3_BUCKET=BUCKET -p 8000:8000 mirecekd/novareel-mcp-server:sse"
echo ""
echo " SSE (With session token):"
echo " docker run -e AWS_ACCESS_KEY_ID=KEY -e AWS_SECRET_ACCESS_KEY=SECRET -e AWS_SESSION_TOKEN=TOKEN -e S3_BUCKET=BUCKET -p 8000:8000 mirecekd/novareel-mcp-server:sse"
echo ""
echo " SSE (With AWS profile):"
echo " docker run -v ~/.aws:/root/.aws -e AWS_PROFILE=my-profile -e S3_BUCKET=BUCKET -p 8000:8000 mirecekd/novareel-mcp-server:sse"
echo ""
echo " HTTP (Environment variables):"
echo " docker run -e AWS_ACCESS_KEY_ID=KEY -e AWS_SECRET_ACCESS_KEY=SECRET -e S3_BUCKET=BUCKET -p 8001:8001 mirecekd/novareel-mcp-server:http"
echo ""
echo " HTTP (With session token):"
echo " docker run -e AWS_ACCESS_KEY_ID=KEY -e AWS_SECRET_ACCESS_KEY=SECRET -e AWS_SESSION_TOKEN=TOKEN -e S3_BUCKET=BUCKET -p 8001:8001 mirecekd/novareel-mcp-server:http"
echo ""
echo " HTTP (With AWS profile):"
echo " docker run -v ~/.aws:/root/.aws -e AWS_PROFILE=my-profile -e S3_BUCKET=BUCKET -p 8001:8001 mirecekd/novareel-mcp-server:http"
```
--------------------------------------------------------------------------------
/.github/workflows/docker-build.yml:
--------------------------------------------------------------------------------
```yaml
name: Build and Push Multi-Arch Docker Images
on:
push:
branches: [ main ]
tags: [ '*' ]
pull_request:
branches: [ main ]
env:
REGISTRY: ghcr.io
IMAGE_NAME: ${{ github.repository }}
jobs:
build:
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
strategy:
matrix:
variant: [stdio, sse, http]
steps:
- name: Checkout repository
uses: actions/checkout@v4
- name: Set up QEMU
uses: docker/setup-qemu-action@v3
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to Container Registry
if: github.event_name != 'pull_request'
uses: docker/login-action@v3
with:
registry: ${{ env.REGISTRY }}
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Extract metadata (tags, labels) for Docker
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
tags: |
type=ref,event=branch,suffix=-${{ matrix.variant }}
type=ref,event=pr,suffix=-${{ matrix.variant }}
type=semver,pattern={{version}},suffix=-${{ matrix.variant }}
type=semver,pattern={{major}}.{{minor}},suffix=-${{ matrix.variant }}
type=semver,pattern={{major}},suffix=-${{ matrix.variant }}
type=raw,value=latest,suffix=-${{ matrix.variant }},enable={{is_default_branch}}
- name: Build and push Docker image
uses: docker/build-push-action@v5
with:
context: .
file: ./Dockerfile.${{ matrix.variant }}
platforms: linux/amd64,linux/arm64,linux/aarch64
push: ${{ github.event_name != 'pull_request' }}
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
cache-from: type=gha
cache-to: type=gha,mode=max
build-summary:
runs-on: ubuntu-latest
needs: build
if: always()
steps:
- name: Build Summary
run: |
echo "## Build Results" >> $GITHUB_STEP_SUMMARY
echo "| Variant | Status |" >> $GITHUB_STEP_SUMMARY
echo "|---------|--------|" >> $GITHUB_STEP_SUMMARY
echo "| stdio | ${{ needs.build.result }} |" >> $GITHUB_STEP_SUMMARY
echo "| sse | ${{ needs.build.result }} |" >> $GITHUB_STEP_SUMMARY
echo "| http | ${{ needs.build.result }} |" >> $GITHUB_STEP_SUMMARY
echo "" >> $GITHUB_STEP_SUMMARY
echo "Images built for platforms: linux/amd64, linux/arm64, linux/aarch64" >> $GITHUB_STEP_SUMMARY
echo "Registry: ghcr.io/${{ github.repository }}" >> $GITHUB_STEP_SUMMARY
echo "" >> $GITHUB_STEP_SUMMARY
echo "Available images:" >> $GITHUB_STEP_SUMMARY
echo "- ghcr.io/${{ github.repository }}:latest-stdio" >> $GITHUB_STEP_SUMMARY
echo "- ghcr.io/${{ github.repository }}:latest-sse" >> $GITHUB_STEP_SUMMARY
echo "- ghcr.io/${{ github.repository }}:latest-http" >> $GITHUB_STEP_SUMMARY
```
--------------------------------------------------------------------------------
/docker-compose.yml:
--------------------------------------------------------------------------------
```yaml
version: '3.8'
services:
novareel-stdio:
image: ghcr.io/mirecekd/novareel-mcp:latest-stdio
# Uncomment to build locally instead:
# build:
# context: .
# dockerfile: Dockerfile.stdio
container_name: novareel-mcp-stdio
environment:
- AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
- AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
- AWS_REGION=${AWS_REGION:-us-east-1}
- S3_BUCKET=${S3_BUCKET}
volumes:
- novareel-data:/root
stdin_open: true
tty: true
restart: unless-stopped
networks:
- novareel-network
novareel-sse:
image: ghcr.io/mirecekd/novareel-mcp:latest-sse
# Uncomment to build locally instead:
# build:
# context: .
# dockerfile: Dockerfile.sse
container_name: novareel-mcp-sse
environment:
- AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
- AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
- AWS_REGION=${AWS_REGION:-us-east-1}
- S3_BUCKET=${S3_BUCKET}
volumes:
- novareel-data:/root
ports:
- "8000:8000"
restart: unless-stopped
networks:
- novareel-network
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
novareel-http:
image: ghcr.io/mirecekd/novareel-mcp:latest-http
# Uncomment to build locally instead:
# build:
# context: .
# dockerfile: Dockerfile.http
container_name: novareel-mcp-http
environment:
- AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
- AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
- AWS_REGION=${AWS_REGION:-us-east-1}
- S3_BUCKET=${S3_BUCKET}
volumes:
- novareel-data:/root
ports:
- "8001:8001"
restart: unless-stopped
networks:
- novareel-network
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8001/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 40s
# Development version with live reloading
novareel-dev:
build:
context: .
dockerfile: Dockerfile.sse
image: mirecekd/novareel-mcp-server:dev
container_name: novareel-mcp-dev
environment:
- AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
- AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
- AWS_REGION=${AWS_REGION:-us-east-1}
- S3_BUCKET=${S3_BUCKET}
ports:
- "8000:8000"
volumes:
- .:/app
restart: unless-stopped
networks:
- novareel-network
profiles:
- dev
volumes:
novareel-data:
driver: local
networks:
novareel-network:
driver: bridge
# Example usage:
# 1. Create .env file with your AWS credentials:
# AWS_ACCESS_KEY_ID=your_access_key
# AWS_SECRET_ACCESS_KEY=your_secret_key
# AWS_REGION=us-east-1
# S3_BUCKET=your-bucket-name
#
# 2. Start both services:
# docker-compose up -d
#
# 3. Use stdio version:
# docker exec -it novareel-mcp-stdio python main.py
#
# 4. Use SSE version:
# Access http://localhost:8000 for web interface
#
# 5. Use HTTP Streaming version:
# Access http://localhost:8001 for HTTP streaming transport
#
# 6. Development with live reloading:
# docker-compose --profile dev up novareel-dev
#
# 7. Stop services:
# docker-compose down
```
--------------------------------------------------------------------------------
/start.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# Nova Reel MCP Server Quick Start Script
set -e
echo "🎬 Amazon Nova Reel MCP Server Quick Start"
echo "=========================================="
# Check if .env file exists
if [ ! -f .env ]; then
echo "⚠️ No .env file found!"
echo "Please copy .env.example to .env and configure your AWS credentials:"
echo ""
echo " cp .env.example .env"
echo " # Edit .env with your AWS credentials"
echo ""
exit 1
fi
# Source environment variables
echo "📋 Loading environment variables..."
export $(cat .env | grep -v '^#' | xargs)
# Check required variables
if [ -z "$S3_BUCKET" ]; then
echo "❌ Missing required S3_BUCKET environment variable!"
echo "Please ensure .env contains S3_BUCKET"
exit 1
fi
# Check if we have valid credential configuration
if [ -n "$AWS_PROFILE" ]; then
echo "✅ Using AWS Profile: $AWS_PROFILE"
elif [ -n "$AWS_ACCESS_KEY_ID" ] && [ -n "$AWS_SECRET_ACCESS_KEY" ]; then
echo "✅ Using explicit AWS credentials"
if [ -n "$AWS_SESSION_TOKEN" ]; then
echo " (with session token for temporary credentials)"
fi
else
echo "❌ Missing AWS credentials configuration!"
echo "Please ensure .env contains either:"
echo " Option 1: AWS_PROFILE=your-profile-name"
echo " Option 2: AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY"
echo " (optionally with AWS_SESSION_TOKEN for temporary credentials)"
exit 1
fi
echo "✅ Environment configured:"
echo " AWS Region: ${AWS_REGION:-us-east-1}"
echo " S3 Bucket: $S3_BUCKET"
echo ""
# Check if we should run stdio or sse version
MODE=${1:-stdio}
case $MODE in
stdio)
echo "🚀 Starting Nova Reel MCP Server (STDIO mode)..."
echo " This mode is for direct MCP client connections."
echo ""
# Build command with conditional parameters
CMD="python main.py --aws-region ${AWS_REGION:-us-east-1} --s3-bucket $S3_BUCKET"
if [ -n "$AWS_PROFILE" ]; then
CMD="$CMD --aws-profile $AWS_PROFILE"
elif [ -n "$AWS_ACCESS_KEY_ID" ] && [ -n "$AWS_SECRET_ACCESS_KEY" ]; then
CMD="$CMD --aws-access-key-id $AWS_ACCESS_KEY_ID --aws-secret-access-key $AWS_SECRET_ACCESS_KEY"
if [ -n "$AWS_SESSION_TOKEN" ]; then
CMD="$CMD --aws-session-token $AWS_SESSION_TOKEN"
fi
fi
eval $CMD
;;
sse)
echo "🚀 Starting Nova Reel MCP Server (SSE mode)..."
echo " This mode provides a web interface."
echo " Access: http://localhost:8000"
echo ""
# Build command with conditional parameters
CMD="python -m novareel_mcp_server.server_sse --aws-region ${AWS_REGION:-us-east-1} --s3-bucket $S3_BUCKET --host 0.0.0.0 --port 8000"
if [ -n "$AWS_PROFILE" ]; then
CMD="$CMD --aws-profile $AWS_PROFILE"
elif [ -n "$AWS_ACCESS_KEY_ID" ] && [ -n "$AWS_SECRET_ACCESS_KEY" ]; then
CMD="$CMD --aws-access-key-id $AWS_ACCESS_KEY_ID --aws-secret-access-key $AWS_SECRET_ACCESS_KEY"
if [ -n "$AWS_SESSION_TOKEN" ]; then
CMD="$CMD --aws-session-token $AWS_SESSION_TOKEN"
fi
fi
eval $CMD
;;
http)
echo "🚀 Starting Nova Reel MCP Server (HTTP Streaming mode)..."
echo " This mode provides HTTP streaming transport."
echo " Access: http://localhost:8001"
echo ""
# Build command with conditional parameters
CMD="python -m novareel_mcp_server.server_http --aws-region ${AWS_REGION:-us-east-1} --s3-bucket $S3_BUCKET --host 0.0.0.0 --port 8001"
if [ -n "$AWS_PROFILE" ]; then
CMD="$CMD --aws-profile $AWS_PROFILE"
elif [ -n "$AWS_ACCESS_KEY_ID" ] && [ -n "$AWS_SECRET_ACCESS_KEY" ]; then
CMD="$CMD --aws-access-key-id $AWS_ACCESS_KEY_ID --aws-secret-access-key $AWS_SECRET_ACCESS_KEY"
if [ -n "$AWS_SESSION_TOKEN" ]; then
CMD="$CMD --aws-session-token $AWS_SESSION_TOKEN"
fi
fi
eval $CMD
;;
docker-stdio)
echo "🐳 Starting Nova Reel MCP Server (Docker STDIO)..."
docker-compose up novareel-stdio
;;
docker-sse)
echo "🐳 Starting Nova Reel MCP Server (Docker SSE)..."
echo " Access: http://localhost:8000"
docker-compose up novareel-sse
;;
docker-http)
echo "🐳 Starting Nova Reel MCP Server (Docker HTTP Streaming)..."
echo " Access: http://localhost:8001"
docker-compose up novareel-http
;;
docker-all)
echo "🐳 Starting all Nova Reel MCP Servers (Docker)..."
echo " SSE Access: http://localhost:8000"
echo " HTTP Access: http://localhost:8001"
docker-compose up -d
echo "✅ All servers started in background"
echo " Use 'docker-compose logs -f' to view logs"
echo " Use 'docker-compose down' to stop"
;;
docker-both)
echo "🐳 Starting both Nova Reel MCP Servers (Docker - legacy)..."
echo " SSE Access: http://localhost:8000"
docker-compose up -d novareel-stdio novareel-sse
echo "✅ Both servers started in background"
echo " Use 'docker-compose logs -f' to view logs"
echo " Use 'docker-compose down' to stop"
;;
build)
echo "🔨 Building all Docker images..."
./build-all.sh
;;
build-stdio)
echo "🔨 Building STDIO Docker image..."
./build-stdio.sh
;;
build-sse)
echo "🔨 Building SSE Docker image..."
./build-sse.sh
;;
build-http)
echo "🔨 Building HTTP Streaming Docker image..."
./build-http.sh
;;
build-package)
echo "🔨 Building Python package..."
./build.sh
;;
*)
echo "❌ Invalid mode: $MODE"
echo ""
echo "Usage: $0 [mode]"
echo ""
echo "Available modes:"
echo " stdio - Run STDIO version locally (default)"
echo " sse - Run SSE version locally"
echo " http - Run HTTP Streaming version locally"
echo " docker-stdio - Run STDIO version in Docker"
echo " docker-sse - Run SSE version in Docker"
echo " docker-http - Run HTTP Streaming version in Docker"
echo " docker-both - Run STDIO + SSE versions in Docker (legacy)"
echo " docker-all - Run all three versions in Docker"
echo " build - Build all Docker images"
echo " build-stdio - Build STDIO Docker image"
echo " build-sse - Build SSE Docker image"
echo " build-http - Build HTTP Streaming Docker image"
echo " build-package - Build Python package (wheel)"
echo ""
echo "Examples:"
echo " $0 # Run STDIO version locally"
echo " $0 sse # Run SSE version locally"
echo " $0 http # Run HTTP Streaming version locally"
echo " $0 build-package # Build Python wheel for uvx"
echo " $0 build # Build all Docker images"
echo " $0 docker-all # Run all three versions in Docker"
exit 1
;;
esac
```
--------------------------------------------------------------------------------
/examples/basic_usage.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Basic usage example for Nova Reel MCP Server
This example demonstrates how to generate a simple video using the MCP server.
"""
import asyncio
import json
import sys
import os
# Add parent directory to path to import the server modules
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from main import start_async_invoke, get_async_invoke, list_async_invokes, get_prompting_guide
async def basic_video_generation():
"""
Example of basic video generation workflow
"""
print("🎬 Nova Reel MCP Server - Basic Usage Example")
print("=" * 50)
# Example prompt for a nature scene
prompt = """
A majestic eagle soars over a mountain valley at golden hour,
camera tracking its flight as it circles above a pristine lake,
then dives gracefully toward the water surface
"""
print(f"📝 Prompt: {prompt.strip()}")
print()
try:
# Start video generation
print("🚀 Starting video generation...")
result = await start_async_invoke(
prompt=prompt,
duration_seconds=24, # 24 second video
fps=24,
dimension="1920x1080" # Full HD
)
if "error" in result:
print(f"❌ Error: {result['error']}")
return
job_id = result["job_id"]
print(f"✅ Job started successfully!")
print(f" Job ID: {job_id}")
print(f" Status: {result['status']}")
print(f" Estimated URL: {result['estimated_video_url']}")
print()
# Monitor progress
print("⏳ Monitoring progress...")
max_attempts = 60 # Wait up to 5 minutes (60 * 5 seconds)
attempt = 0
while attempt < max_attempts:
status_result = await get_async_invoke(job_id)
if "error" in status_result:
print(f"❌ Error checking status: {status_result['error']}")
break
current_status = status_result["status"]
print(f" Status: {current_status} (attempt {attempt + 1}/{max_attempts})")
if current_status == "Completed":
print("🎉 Video generation completed!")
print(f" Video URL: {status_result['video_url']}")
print(f" Duration: {status_result['config']['duration_seconds']} seconds")
print(f" Dimensions: {status_result['config']['dimension']}")
break
elif current_status in ["Failed", "Cancelled"]:
print(f"❌ Video generation {current_status.lower()}")
if "failure_message" in status_result:
print(f" Reason: {status_result['failure_message']}")
break
# Wait 5 seconds before checking again
await asyncio.sleep(5)
attempt += 1
if attempt >= max_attempts:
print("⏰ Timeout reached. Video may still be processing.")
print(" Use get_async_invoke() to check status later.")
except Exception as e:
print(f"❌ Unexpected error: {e}")
async def list_all_jobs():
"""
Example of listing all video generation jobs
"""
print("\n📋 Listing all jobs...")
print("-" * 30)
try:
jobs_result = await list_async_invokes()
if "error" in jobs_result:
print(f"❌ Error: {jobs_result['error']}")
return
total = jobs_result["total_invocations"]
summary = jobs_result["summary"]
print(f"Total jobs: {total}")
print(f" ✅ Completed: {summary['completed']}")
print(f" ⏳ In Progress: {summary['in_progress']}")
print(f" ❌ Failed: {summary['failed']}")
print(f" ❓ Unknown: {summary['unknown']}")
print()
if total > 0:
print("Recent jobs:")
for job in jobs_result["invocations"][:5]: # Show last 5 jobs
status_emoji = {
"Completed": "✅",
"InProgress": "⏳",
"Failed": "❌",
"Cancelled": "❌",
"Unknown": "❓"
}.get(job["status"], "❓")
print(f" {status_emoji} {job['job_id'][:8]}... - {job['status']}")
print(f" Prompt: {job['prompt'][:60]}...")
if job.get("video_url"):
print(f" URL: {job['video_url']}")
print()
except Exception as e:
print(f"❌ Unexpected error: {e}")
async def show_prompting_tips():
"""
Example of getting prompting guidelines
"""
print("\n💡 Prompting Guidelines")
print("=" * 30)
try:
guide = await get_prompting_guide()
# Show basic principles
print("Basic Principles:")
for principle, details in guide["basic_principles"].items():
print(f"\n🔹 {principle.replace('_', ' ').title()}:")
print(f" {details['description']}")
print(f" ✅ Good: {details['good_example']}")
print(f" ❌ Bad: {details['bad_example']}")
# Show example prompts
print("\n\n📝 Example Prompts:")
examples = guide["example_prompts"]
for category, prompts in examples.items():
print(f"\n🎯 {category.title()}:")
for duration, prompt in prompts.items():
print(f" {duration.replace('_', ' ').title()}: {prompt}")
# Show common mistakes
print("\n\n⚠️ Common Mistakes to Avoid:")
for mistake, details in guide["common_mistakes"].items():
print(f"\n❌ {details['problem']}")
print(f" Example: {details['example']}")
print(f" Solution: {details['solution']}")
except Exception as e:
print(f"❌ Unexpected error: {e}")
async def main():
"""
Main example function
"""
print("Welcome to Nova Reel MCP Server Examples!")
print("This example will demonstrate basic video generation.")
print()
# Check if AWS credentials are configured
if not all([
os.getenv("AWS_ACCESS_KEY_ID"),
os.getenv("AWS_SECRET_ACCESS_KEY"),
os.getenv("S3_BUCKET")
]):
print("⚠️ AWS credentials not configured!")
print("Please set the following environment variables:")
print(" - AWS_ACCESS_KEY_ID")
print(" - AWS_SECRET_ACCESS_KEY")
print(" - S3_BUCKET")
print(" - AWS_REGION (optional, defaults to us-east-1)")
print()
print("Example:")
print(" export AWS_ACCESS_KEY_ID=your_access_key")
print(" export AWS_SECRET_ACCESS_KEY=your_secret_key")
print(" export S3_BUCKET=your-bucket-name")
print(" python examples/basic_usage.py")
return
# Show prompting tips first
await show_prompting_tips()
# Generate a video
await basic_video_generation()
# List all jobs
await list_all_jobs()
print("\n🎉 Example completed!")
print("Check your S3 bucket for the generated video.")
if __name__ == "__main__":
# Note: This example assumes the MCP server functions are available
# In a real scenario, you would interact with the MCP server via the protocol
print("📝 Note: This is a conceptual example.")
print("In practice, you would interact with the MCP server through an MCP client.")
print("This example shows the expected workflow and API usage.")
# Run the example
asyncio.run(main())
```
--------------------------------------------------------------------------------
/src/novareel_mcp_server/prompting_guide.py:
--------------------------------------------------------------------------------
```python
"""
Amazon Nova Reel Prompting Guidelines
Based on AWS documentation for video generation and camera control.
"""
def get_prompting_guidelines():
"""
Returns comprehensive prompting guidelines for Amazon Nova Reel video generation.
Based on AWS documentation:
- https://docs.aws.amazon.com/nova/latest/userguide/prompting-video-generation.html
- https://docs.aws.amazon.com/nova/latest/userguide/prompting-video-camera-control.html
"""
return {
"overview": {
"title": "Amazon Nova Reel Video Generation Prompting Guide",
"description": "Best practices for creating effective prompts for video generation with Amazon Nova Reel",
"model": "amazon.nova-reel-v1:1",
"supported_durations": "12-120 seconds (multiples of 6)",
"supported_dimensions": ["1280x720", "1920x1080", "1024x1024"]
},
"basic_principles": {
"be_specific": {
"description": "Use specific, descriptive language rather than vague terms",
"good_example": "A red cardinal perched on a snow-covered pine branch, morning sunlight filtering through the trees",
"bad_example": "A bird on a tree"
},
"use_active_language": {
"description": "Use active voice and present tense for dynamic scenes",
"good_example": "The waves crash against the rocky shore as seagulls soar overhead",
"bad_example": "Waves were crashing and birds were flying"
},
"include_context": {
"description": "Provide environmental and atmospheric details",
"good_example": "In a bustling Tokyo street at night, neon signs reflect on wet pavement as people hurry past",
"bad_example": "People walking in a city"
}
},
"video_structure": {
"beginning_middle_end": {
"description": "Structure your prompt with a clear progression",
"example": "A butterfly lands on a flower (beginning), slowly opens and closes its wings (middle), then flies away into the sunset (end)",
"tip": "For longer videos, describe multiple scenes or actions in sequence"
},
"pacing": {
"description": "Consider the pacing of actions for your video duration",
"short_videos": "Focus on single actions or moments (12-24 seconds)",
"medium_videos": "Include 2-3 distinct actions or scene changes (30-60 seconds)",
"long_videos": "Develop a narrative with multiple scenes (60-120 seconds)"
}
},
"camera_control": {
"overview": "Use specific camera terminology to control shot composition and movement",
"shot_types": {
"close_up": "Close-up shot of a person's face showing detailed expressions",
"medium_shot": "Medium shot showing a person from waist up",
"wide_shot": "Wide shot establishing the entire scene and environment",
"extreme_close_up": "Extreme close-up focusing on eyes or hands",
"establishing_shot": "Establishing shot revealing the location and setting"
},
"camera_movements": {
"pan": "Camera pans left/right across the landscape",
"tilt": "Camera tilts up to reveal the towering mountain",
"zoom": "Camera slowly zooms in on the subject's face",
"dolly": "Camera dollies forward through the forest path",
"tracking": "Camera tracks alongside the running athlete",
"crane": "Camera cranes up to show the aerial view of the city"
},
"angles": {
"low_angle": "Low angle shot looking up at the imposing building",
"high_angle": "High angle shot looking down at the busy street",
"bird_eye": "Bird's eye view of the circular plaza",
"worm_eye": "Worm's eye view of the towering trees",
"dutch_angle": "Dutch angle creating a sense of unease"
},
"depth_of_field": {
"shallow": "Shallow depth of field with blurred background",
"deep": "Deep focus keeping both foreground and background sharp",
"rack_focus": "Rack focus shifting from foreground to background"
}
},
"lighting_and_atmosphere": {
"natural_lighting": {
"golden_hour": "Warm golden hour lighting casting long shadows",
"blue_hour": "Soft blue hour twilight with city lights beginning to glow",
"harsh_sunlight": "Bright midday sun creating strong contrasts",
"overcast": "Soft, diffused lighting from overcast sky"
},
"artificial_lighting": {
"neon": "Colorful neon lights reflecting on wet streets",
"candlelight": "Warm, flickering candlelight creating intimate atmosphere",
"spotlight": "Dramatic spotlight illuminating the performer",
"backlighting": "Strong backlighting creating silhouettes"
},
"weather_atmosphere": {
"fog": "Mysterious fog rolling through the valley",
"rain": "Heavy rain creating ripples in puddles",
"snow": "Gentle snowfall in the quiet forest",
"storm": "Dramatic storm clouds gathering overhead"
}
},
"subject_and_action": {
"people": {
"emotions": "Include specific emotions and expressions",
"clothing": "Describe clothing style and colors",
"age_appearance": "Specify age range and general appearance",
"actions": "Use specific action verbs (strolling, sprinting, gesturing)"
},
"animals": {
"species": "Be specific about animal species and breeds",
"behavior": "Describe natural behaviors and movements",
"habitat": "Include appropriate natural habitat details"
},
"objects": {
"materials": "Specify materials (wooden, metallic, glass, fabric)",
"condition": "Describe condition (new, weathered, antique, modern)",
"interaction": "How objects interact with environment or subjects"
}
},
"style_and_genre": {
"cinematic_styles": {
"documentary": "Documentary style with natural, observational camera work",
"commercial": "Polished commercial style with perfect lighting",
"indie_film": "Indie film aesthetic with handheld camera movement",
"music_video": "Dynamic music video style with quick cuts and effects"
},
"visual_styles": {
"realistic": "Photorealistic style with natural colors and lighting",
"stylized": "Stylized with enhanced colors and dramatic lighting",
"vintage": "Vintage film look with grain and muted colors",
"modern": "Clean, modern aesthetic with sharp details"
}
},
"technical_considerations": {
"frame_rate": {
"24fps": "Standard cinematic frame rate for natural motion",
"higher_fps": "Higher frame rates for smooth slow-motion effects"
},
"resolution": {
"1280x720": "HD resolution suitable for most applications",
"1920x1080": "Full HD for higher quality output",
"1024x1024": "Square format for social media"
},
"duration_planning": {
"12_seconds": "Perfect for single action or moment",
"24_seconds": "Good for simple scene with beginning and end",
"60_seconds": "Allows for multiple actions or scene progression",
"120_seconds": "Full narrative with multiple scenes possible"
}
},
"example_prompts": {
"nature": {
"short": "Close-up of morning dew drops on a spider web, with soft sunrise lighting creating rainbow reflections",
"medium": "A majestic eagle soars over a mountain valley, camera tracking its flight as it circles above a pristine lake, then dives toward the water",
"long": "Time-lapse of a flower blooming in a meadow: starting as a bud at dawn, slowly opening petals as the sun rises, bees visiting throughout the day, and closing as sunset approaches"
},
"urban": {
"short": "Neon signs reflecting in rain puddles on a busy Tokyo street at night, with people's feet splashing through the colorful reflections",
"medium": "A street musician plays violin in a subway station, commuters pause to listen, coins drop into his case, camera slowly pulls back to reveal the bustling underground scene",
"long": "Morning rush hour in Manhattan: alarm clocks ring, people emerge from apartments, flood the sidewalks, enter subway stations, trains arrive and depart, finally arriving at office buildings as the city comes alive"
},
"portrait": {
"short": "Extreme close-up of an elderly craftsman's weathered hands carving intricate details into wood, with warm workshop lighting",
"medium": "A young dancer practices alone in a sunlit studio, her movements flowing from gentle stretches to powerful leaps, shadows dancing on the wooden floor",
"long": "Portrait of a chef preparing a signature dish: selecting fresh ingredients at market, returning to kitchen, methodically preparing each component, plating with artistic precision, and presenting the finished masterpiece"
}
},
"common_mistakes": {
"too_vague": {
"problem": "Prompts that are too general or vague",
"example": "A person doing something",
"solution": "Be specific about who, what, where, when, and how"
},
"conflicting_elements": {
"problem": "Including contradictory or impossible elements",
"example": "Underwater scene with fire burning",
"solution": "Ensure all elements are physically and logically consistent"
},
"overcomplication": {
"problem": "Trying to include too many elements or actions",
"example": "A person cooking while dancing while painting while talking on phone",
"solution": "Focus on 1-3 main elements or actions for clarity"
},
"inappropriate_duration": {
"problem": "Describing actions that don't match video duration",
"example": "Describing a 5-minute cooking process for a 12-second video",
"solution": "Match action complexity to video duration"
}
},
"optimization_tips": {
"use_keywords": "Include relevant keywords for style, mood, and technical aspects",
"specify_quality": "Add terms like 'high quality', 'detailed', 'professional' for better results",
"mention_equipment": "Reference camera types or lenses for specific looks (e.g., 'shot with 85mm lens')",
"include_mood": "Describe the emotional tone or atmosphere you want to convey",
"test_variations": "Try different phrasings of the same concept to find what works best"
},
"prompt_templates": {
"basic_template": "[Subject] [Action] [Location] [Lighting] [Camera angle] [Style]",
"narrative_template": "[Opening scene], [transition/development], [conclusion/resolution]",
"technical_template": "[Shot type] of [subject] [action] in [environment], [lighting], [camera movement], [style]"
}
}
```
--------------------------------------------------------------------------------
/src/novareel_mcp_server/server_sse.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Amazon Nova Reel 1.1 MCP Server - SSE Version
Provides tools for video generation using AWS Bedrock Nova Reel model via Server-Sent Events.
"""
import argparse
import asyncio
import json
import os
import sys
import random
import time
from datetime import datetime
from typing import Optional, Dict, Any, List
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
from fastmcp import FastMCP
from .prompting_guide import get_prompting_guidelines
# Create MCP server with SSE transport
mcp = FastMCP("Amazon Nova Reel 1.1 SSE")
# Global variables for AWS configuration
aws_access_key_id: Optional[str] = None
aws_secret_access_key: Optional[str] = None
aws_session_token: Optional[str] = None
aws_profile: Optional[str] = None
aws_region: Optional[str] = None
s3_bucket: Optional[str] = None
bedrock_client = None
# Model configuration
MODEL_ID = "amazon.nova-reel-v1:1"
SLEEP_SECONDS = 5 # Interval for checking video generation progress
# In-memory storage for tracking invocations (in production, use persistent storage)
active_invocations = {}
class NovaReelError(Exception):
"""Base exception for Nova Reel operations"""
pass
class AWSConfigError(NovaReelError):
"""AWS configuration error"""
pass
class VideoGenerationError(NovaReelError):
"""Video generation error"""
pass
def initialize_aws_client():
"""Initialize AWS Bedrock client with provided credentials or profile"""
global bedrock_client
if not s3_bucket:
raise AWSConfigError("Missing required S3_BUCKET configuration")
try:
# Option 1: Use AWS Profile
if aws_profile:
print(f"Using AWS profile: {aws_profile}", file=sys.stderr)
session = boto3.Session(profile_name=aws_profile, region_name=aws_region)
bedrock_client = session.client("bedrock-runtime")
# Option 2: Use explicit credentials
elif aws_access_key_id and aws_secret_access_key:
print("Using explicit AWS credentials", file=sys.stderr)
client_kwargs = {
"service_name": "bedrock-runtime",
"region_name": aws_region,
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key
}
# Add session token if provided (for temporary credentials)
if aws_session_token:
client_kwargs["aws_session_token"] = aws_session_token
print("Using temporary credentials with session token", file=sys.stderr)
bedrock_client = boto3.client(**client_kwargs)
# Option 3: Use default credential chain
else:
print("Using default AWS credential chain", file=sys.stderr)
bedrock_client = boto3.client("bedrock-runtime", region_name=aws_region)
# Test the connection with a simple operation
# Note: bedrock-runtime doesn't have list_foundation_models, that's in bedrock client
# We'll just create the client and let the first actual call test the connection
except NoCredentialsError:
raise AWSConfigError("No valid AWS credentials found. Please provide explicit credentials, set AWS_PROFILE, or configure default credentials.")
except ClientError as e:
raise AWSConfigError(f"AWS client error: {e}")
except Exception as e:
raise AWSConfigError(f"Failed to initialize AWS client: {e}")
@mcp.tool()
async def start_async_invoke(
prompt: str,
duration_seconds: int = 12,
fps: int = 24,
dimension: str = "1280x720",
seed: Optional[int] = None,
task_type: str = "MULTI_SHOT_AUTOMATED"
) -> Dict[str, Any]:
"""
Start asynchronous video generation with Amazon Nova Reel.
Args:
prompt: Text description for video generation. See prompting guidelines for best practices.
duration_seconds: Video duration in seconds (must be multiple of 6, range 12-120)
fps: Frames per second (24 recommended)
dimension: Video dimensions (1280x720, 1920x1080, etc.)
seed: Random seed for reproducible results (optional)
task_type: Task type (MULTI_SHOT_AUTOMATED recommended)
Returns:
Dict containing invocation details and job information
"""
try:
if not bedrock_client:
initialize_aws_client()
# Validate duration
if duration_seconds < 12 or duration_seconds > 120 or duration_seconds % 6 != 0:
return {
"error": "Duration must be a multiple of 6 in range [12, 120]",
"valid_durations": [12, 18, 24, 30, 36, 42, 48, 54, 60, 66, 72, 78, 84, 90, 96, 102, 108, 114, 120]
}
# Generate seed if not provided
if seed is None:
seed = random.randint(0, 2147483648)
# Prepare model input
model_input = {
"taskType": task_type,
"multiShotAutomatedParams": {"text": prompt},
"videoGenerationConfig": {
"durationSeconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed,
},
}
# Start async invocation
invocation = bedrock_client.start_async_invoke(
modelId=MODEL_ID,
modelInput=model_input,
outputDataConfig={"s3OutputDataConfig": {"s3Uri": f"s3://{s3_bucket}"}},
)
invocation_arn = invocation["invocationArn"]
job_id = invocation_arn.split("/")[-1]
s3_location = f"s3://{s3_bucket}/{job_id}"
# Store invocation details
invocation_data = {
"invocation_arn": invocation_arn,
"job_id": job_id,
"prompt": prompt,
"duration_seconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed,
"task_type": task_type,
"s3_location": s3_location,
"status": "InProgress",
"created_at": datetime.now().isoformat(),
"video_url": None
}
active_invocations[job_id] = invocation_data
return {
"success": True,
"invocation_arn": invocation_arn,
"job_id": job_id,
"status": "InProgress",
"s3_location": s3_location,
"estimated_video_url": f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4",
"prompt": prompt,
"config": {
"duration_seconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed
},
"message": "Video generation started. Use get_async_invoke to check progress."
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except ClientError as e:
return {"error": f"AWS API error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def list_async_invokes() -> Dict[str, Any]:
"""
List all tracked async video generation invocations.
Returns:
Dict containing list of all invocations with their current status
"""
try:
if not bedrock_client:
initialize_aws_client()
# Update status for all active invocations
updated_invocations = []
for job_id, invocation_data in active_invocations.items():
try:
# Get current status from AWS
response = bedrock_client.get_async_invoke(
invocationArn=invocation_data["invocation_arn"]
)
# Update status
current_status = response["status"]
invocation_data["status"] = current_status
if current_status == "Completed":
invocation_data["video_url"] = f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4"
invocation_data["completed_at"] = datetime.now().isoformat()
elif current_status in ["Failed", "Cancelled"]:
invocation_data["failed_at"] = datetime.now().isoformat()
if "failureMessage" in response:
invocation_data["failure_message"] = response["failureMessage"]
updated_invocations.append({
"job_id": job_id,
"status": current_status,
"prompt": invocation_data["prompt"][:100] + "..." if len(invocation_data["prompt"]) > 100 else invocation_data["prompt"],
"created_at": invocation_data["created_at"],
"video_url": invocation_data.get("video_url"),
"duration_seconds": invocation_data["duration_seconds"]
})
except ClientError as e:
# If we can't get status, mark as unknown
invocation_data["status"] = "Unknown"
invocation_data["error"] = str(e)
updated_invocations.append({
"job_id": job_id,
"status": "Unknown",
"prompt": invocation_data["prompt"][:100] + "..." if len(invocation_data["prompt"]) > 100 else invocation_data["prompt"],
"created_at": invocation_data["created_at"],
"error": str(e)
})
return {
"success": True,
"total_invocations": len(updated_invocations),
"invocations": updated_invocations,
"summary": {
"in_progress": len([inv for inv in updated_invocations if inv["status"] == "InProgress"]),
"completed": len([inv for inv in updated_invocations if inv["status"] == "Completed"]),
"failed": len([inv for inv in updated_invocations if inv["status"] in ["Failed", "Cancelled"]]),
"unknown": len([inv for inv in updated_invocations if inv["status"] == "Unknown"])
}
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def get_async_invoke(identifier: str) -> Dict[str, Any]:
"""
Get detailed information about a specific async video generation invocation.
Args:
identifier: Either job_id or invocation_arn
Returns:
Dict containing detailed invocation information and video URL if completed
"""
try:
if not bedrock_client:
initialize_aws_client()
# Find invocation by job_id or invocation_arn
invocation_data = None
job_id = None
if identifier in active_invocations:
# Direct job_id lookup
job_id = identifier
invocation_data = active_invocations[identifier]
else:
# Search by invocation_arn
for jid, data in active_invocations.items():
if data["invocation_arn"] == identifier:
job_id = jid
invocation_data = data
break
if not invocation_data:
return {
"error": f"Invocation not found: {identifier}",
"suggestion": "Use list_async_invokes to see all tracked invocations"
}
# Get current status from AWS
try:
response = bedrock_client.get_async_invoke(
invocationArn=invocation_data["invocation_arn"]
)
current_status = response["status"]
invocation_data["status"] = current_status
# Prepare detailed response
result = {
"success": True,
"job_id": job_id,
"invocation_arn": invocation_data["invocation_arn"],
"status": current_status,
"prompt": invocation_data["prompt"],
"config": {
"duration_seconds": invocation_data["duration_seconds"],
"fps": invocation_data["fps"],
"dimension": invocation_data["dimension"],
"seed": invocation_data["seed"],
"task_type": invocation_data["task_type"]
},
"s3_location": invocation_data["s3_location"],
"created_at": invocation_data["created_at"]
}
if current_status == "Completed":
video_url = f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4"
invocation_data["video_url"] = video_url
invocation_data["completed_at"] = datetime.now().isoformat()
result["video_url"] = video_url
result["completed_at"] = invocation_data["completed_at"]
result["message"] = "Video generation completed successfully!"
elif current_status == "InProgress":
result["message"] = "Video generation is still in progress. Check again in a few moments."
elif current_status in ["Failed", "Cancelled"]:
invocation_data["failed_at"] = datetime.now().isoformat()
result["failed_at"] = invocation_data["failed_at"]
result["message"] = f"Video generation {current_status.lower()}"
if "failureMessage" in response:
result["failure_message"] = response["failureMessage"]
invocation_data["failure_message"] = response["failureMessage"]
return result
except ClientError as e:
return {
"error": f"Failed to get invocation status: {e}",
"job_id": job_id,
"last_known_status": invocation_data.get("status", "Unknown")
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def get_prompting_guide() -> Dict[str, Any]:
"""
Get comprehensive prompting guidelines for Amazon Nova Reel video generation.
Returns:
Dict containing prompting best practices and examples
"""
return get_prompting_guidelines()
def main():
"""Main function to run the MCP server with SSE transport"""
parser = argparse.ArgumentParser(description="Amazon Nova Reel 1.1 MCP Server - SSE Version")
parser.add_argument("--aws-access-key-id", help="AWS Access Key ID")
parser.add_argument("--aws-secret-access-key", help="AWS Secret Access Key")
parser.add_argument("--aws-session-token", help="AWS Session Token (for temporary credentials)")
parser.add_argument("--aws-profile", help="AWS Profile name (alternative to explicit credentials)")
parser.add_argument("--aws-region", default="us-east-1", help="AWS Region")
parser.add_argument("--s3-bucket", help="S3 bucket name for video output")
parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
parser.add_argument("--port", type=int, default=8000, help="Port to bind to")
args = parser.parse_args()
# Set global configuration from args or environment variables
global aws_access_key_id, aws_secret_access_key, aws_session_token, aws_profile, aws_region, s3_bucket
aws_access_key_id = args.aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID")
aws_secret_access_key = args.aws_secret_access_key or os.getenv("AWS_SECRET_ACCESS_KEY")
aws_session_token = args.aws_session_token or os.getenv("AWS_SESSION_TOKEN")
aws_profile = args.aws_profile or os.getenv("AWS_PROFILE")
aws_region = args.aws_region or os.getenv("AWS_REGION", "us-east-1")
s3_bucket = args.s3_bucket or os.getenv("S3_BUCKET")
# Validate configuration - need either profile OR explicit credentials + S3 bucket
if not s3_bucket:
print("Error: Missing required S3_BUCKET configuration.", file=sys.stderr)
print("Please provide --s3-bucket or S3_BUCKET env var", file=sys.stderr)
sys.exit(1)
# Check if we have valid credential configuration
has_explicit_creds = aws_access_key_id and aws_secret_access_key
has_profile = aws_profile
if not has_explicit_creds and not has_profile:
print("Error: Missing AWS credentials configuration.", file=sys.stderr)
print("Please provide either:", file=sys.stderr)
print(" Option 1: --aws-access-key-id and --aws-secret-access-key (with optional --aws-session-token)", file=sys.stderr)
print(" Option 2: --aws-profile", file=sys.stderr)
print(" Option 3: Set corresponding environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_PROFILE)", file=sys.stderr)
print(" Option 4: Configure default AWS credentials (e.g., via aws configure)", file=sys.stderr)
sys.exit(1)
if has_explicit_creds and has_profile:
print("Warning: Both explicit credentials and AWS profile provided. Using explicit credentials.", file=sys.stderr)
aws_profile = None # Clear profile to avoid confusion
# Remove s3:// prefix if present
if s3_bucket.startswith("s3://"):
s3_bucket = s3_bucket[5:]
# Initialize AWS client
try:
initialize_aws_client()
print(f"Nova Reel MCP Server (SSE) initialized with region: {aws_region}, bucket: {s3_bucket}", file=sys.stderr)
print(f"Starting server on {args.host}:{args.port}", file=sys.stderr)
except AWSConfigError as e:
print(f"AWS configuration error: {e}", file=sys.stderr)
sys.exit(1)
# Run MCP server with SSE transport
mcp.run(transport="sse", host=args.host, port=args.port)
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/novareel_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Amazon Nova Reel 1.1 MCP Server
Provides tools for video generation using AWS Bedrock Nova Reel model via Model Context Protocol.
"""
import argparse
import asyncio
import json
import os
import sys
import random
import time
from datetime import datetime
from typing import Optional, Dict, Any, List
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
from fastmcp import FastMCP
from .prompting_guide import get_prompting_guidelines
# Create MCP server
mcp = FastMCP("Amazon Nova Reel 1.1")
# Global variables for AWS configuration
aws_access_key_id: Optional[str] = None
aws_secret_access_key: Optional[str] = None
aws_session_token: Optional[str] = None
aws_profile: Optional[str] = None
aws_region: Optional[str] = None
s3_bucket: Optional[str] = None
bedrock_client = None
# Model configuration
MODEL_ID = "amazon.nova-reel-v1:1"
SLEEP_SECONDS = 5 # Interval for checking video generation progress
# Persistent storage for tracking invocations
INVOCATIONS_FILE = os.path.expanduser("~/.novareel_invocations.json")
active_invocations = {}
def load_invocations():
"""Load invocations from persistent storage"""
global active_invocations
try:
if os.path.exists(INVOCATIONS_FILE):
with open(INVOCATIONS_FILE, 'r') as f:
active_invocations = json.load(f)
except Exception as e:
print(f"Warning: Could not load invocations file: {e}", file=sys.stderr)
active_invocations = {}
def save_invocations():
"""Save invocations to persistent storage"""
try:
with open(INVOCATIONS_FILE, 'w') as f:
json.dump(active_invocations, f, indent=2)
except Exception as e:
print(f"Warning: Could not save invocations file: {e}", file=sys.stderr)
class NovaReelError(Exception):
"""Base exception for Nova Reel operations"""
pass
class AWSConfigError(NovaReelError):
"""AWS configuration error"""
pass
class VideoGenerationError(NovaReelError):
"""Video generation error"""
pass
def initialize_aws_client():
"""Initialize AWS Bedrock client with provided credentials or profile"""
global bedrock_client
if not s3_bucket:
raise AWSConfigError("Missing required S3_BUCKET configuration")
try:
# Option 1: Use AWS Profile
if aws_profile:
print(f"Using AWS profile: {aws_profile}", file=sys.stderr)
session = boto3.Session(profile_name=aws_profile, region_name=aws_region)
bedrock_client = session.client("bedrock-runtime")
# Option 2: Use explicit credentials
elif aws_access_key_id and aws_secret_access_key:
print("Using explicit AWS credentials", file=sys.stderr)
client_kwargs = {
"service_name": "bedrock-runtime",
"region_name": aws_region,
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key
}
# Add session token if provided (for temporary credentials)
if aws_session_token:
client_kwargs["aws_session_token"] = aws_session_token
print("Using temporary credentials with session token", file=sys.stderr)
bedrock_client = boto3.client(**client_kwargs)
# Option 3: Use default credential chain
else:
print("Using default AWS credential chain", file=sys.stderr)
bedrock_client = boto3.client("bedrock-runtime", region_name=aws_region)
# Test the connection with a simple operation
# Note: bedrock-runtime doesn't have list_foundation_models, that's in bedrock client
# We'll just create the client and let the first actual call test the connection
except NoCredentialsError:
raise AWSConfigError("No valid AWS credentials found. Please provide explicit credentials, set AWS_PROFILE, or configure default credentials.")
except ClientError as e:
raise AWSConfigError(f"AWS client error: {e}")
except Exception as e:
raise AWSConfigError(f"Failed to initialize AWS client: {e}")
@mcp.tool()
async def start_async_invoke(
prompt: str,
duration_seconds: int = 12,
fps: int = 24,
dimension: str = "1280x720",
seed: Optional[int] = None,
task_type: str = "MULTI_SHOT_AUTOMATED"
) -> Dict[str, Any]:
"""
Start asynchronous video generation with Amazon Nova Reel.
Args:
prompt: Text description for video generation. See prompting guidelines for best practices.
duration_seconds: Video duration in seconds (must be multiple of 6, range 12-120)
fps: Frames per second (24 recommended)
dimension: Video dimensions (1280x720, 1920x1080, etc.)
seed: Random seed for reproducible results (optional)
task_type: Task type (MULTI_SHOT_AUTOMATED recommended)
Returns:
Dict containing invocation details and job information
"""
try:
if not bedrock_client:
initialize_aws_client()
# Validate duration
if duration_seconds < 12 or duration_seconds > 120 or duration_seconds % 6 != 0:
return {
"error": "Duration must be a multiple of 6 in range [12, 120]",
"valid_durations": [12, 18, 24, 30, 36, 42, 48, 54, 60, 66, 72, 78, 84, 90, 96, 102, 108, 114, 120]
}
# Generate seed if not provided
if seed is None:
seed = random.randint(0, 2147483648)
# Prepare model input
model_input = {
"taskType": task_type,
"multiShotAutomatedParams": {"text": prompt},
"videoGenerationConfig": {
"durationSeconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed,
},
}
# Start async invocation
invocation = bedrock_client.start_async_invoke(
modelId=MODEL_ID,
modelInput=model_input,
outputDataConfig={"s3OutputDataConfig": {"s3Uri": f"s3://{s3_bucket}"}},
)
invocation_arn = invocation["invocationArn"]
job_id = invocation_arn.split("/")[-1]
s3_location = f"s3://{s3_bucket}/{job_id}"
# Store invocation details
invocation_data = {
"invocation_arn": invocation_arn,
"job_id": job_id,
"prompt": prompt,
"duration_seconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed,
"task_type": task_type,
"s3_location": s3_location,
"status": "InProgress",
"created_at": datetime.now().isoformat(),
"video_url": None
}
active_invocations[job_id] = invocation_data
save_invocations() # Save to persistent storage
return {
"success": True,
"invocation_arn": invocation_arn,
"job_id": job_id,
"status": "InProgress",
"s3_location": s3_location,
"estimated_video_url": f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4",
"prompt": prompt,
"config": {
"duration_seconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed
},
"message": "Video generation started. Use get_async_invoke to check progress."
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except ClientError as e:
return {"error": f"AWS API error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def list_async_invokes() -> Dict[str, Any]:
"""
List all tracked async video generation invocations.
Returns:
Dict containing list of all invocations with their current status
"""
try:
if not bedrock_client:
initialize_aws_client()
# Update status for all active invocations
updated_invocations = []
for job_id, invocation_data in active_invocations.items():
try:
# Get current status from AWS
response = bedrock_client.get_async_invoke(
invocationArn=invocation_data["invocation_arn"]
)
# Update status
current_status = response["status"]
invocation_data["status"] = current_status
if current_status == "Completed":
invocation_data["video_url"] = f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4"
invocation_data["completed_at"] = datetime.now().isoformat()
elif current_status in ["Failed", "Cancelled"]:
invocation_data["failed_at"] = datetime.now().isoformat()
if "failureMessage" in response:
invocation_data["failure_message"] = response["failureMessage"]
updated_invocations.append({
"job_id": job_id,
"status": current_status,
"prompt": invocation_data["prompt"][:100] + "..." if len(invocation_data["prompt"]) > 100 else invocation_data["prompt"],
"created_at": invocation_data["created_at"],
"video_url": invocation_data.get("video_url"),
"duration_seconds": invocation_data["duration_seconds"]
})
except ClientError as e:
# If we can't get status, mark as unknown
invocation_data["status"] = "Unknown"
invocation_data["error"] = str(e)
updated_invocations.append({
"job_id": job_id,
"status": "Unknown",
"prompt": invocation_data["prompt"][:100] + "..." if len(invocation_data["prompt"]) > 100 else invocation_data["prompt"],
"created_at": invocation_data["created_at"],
"error": str(e)
})
return {
"success": True,
"total_invocations": len(updated_invocations),
"invocations": updated_invocations,
"summary": {
"in_progress": len([inv for inv in updated_invocations if inv["status"] == "InProgress"]),
"completed": len([inv for inv in updated_invocations if inv["status"] == "Completed"]),
"failed": len([inv for inv in updated_invocations if inv["status"] in ["Failed", "Cancelled"]]),
"unknown": len([inv for inv in updated_invocations if inv["status"] == "Unknown"])
}
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def get_async_invoke(identifier: str) -> Dict[str, Any]:
"""
Get detailed information about a specific async video generation invocation.
Args:
identifier: Either job_id or invocation_arn
Returns:
Dict containing detailed invocation information and video URL if completed
"""
try:
if not bedrock_client:
initialize_aws_client()
# Find invocation by job_id or invocation_arn
invocation_data = None
job_id = None
if identifier in active_invocations:
# Direct job_id lookup
job_id = identifier
invocation_data = active_invocations[identifier]
else:
# Search by invocation_arn
for jid, data in active_invocations.items():
if data["invocation_arn"] == identifier:
job_id = jid
invocation_data = data
break
if not invocation_data:
return {
"error": f"Invocation not found: {identifier}",
"suggestion": "Use list_async_invokes to see all tracked invocations"
}
# Get current status from AWS
try:
response = bedrock_client.get_async_invoke(
invocationArn=invocation_data["invocation_arn"]
)
current_status = response["status"]
invocation_data["status"] = current_status
# Prepare detailed response
result = {
"success": True,
"job_id": job_id,
"invocation_arn": invocation_data["invocation_arn"],
"status": current_status,
"prompt": invocation_data["prompt"],
"config": {
"duration_seconds": invocation_data["duration_seconds"],
"fps": invocation_data["fps"],
"dimension": invocation_data["dimension"],
"seed": invocation_data["seed"],
"task_type": invocation_data["task_type"]
},
"s3_location": invocation_data["s3_location"],
"created_at": invocation_data["created_at"]
}
if current_status == "Completed":
video_url = f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4"
invocation_data["video_url"] = video_url
invocation_data["completed_at"] = datetime.now().isoformat()
result["video_url"] = video_url
result["completed_at"] = invocation_data["completed_at"]
result["message"] = "Video generation completed successfully!"
elif current_status == "InProgress":
result["message"] = "Video generation is still in progress. Check again in a few moments."
elif current_status in ["Failed", "Cancelled"]:
invocation_data["failed_at"] = datetime.now().isoformat()
result["failed_at"] = invocation_data["failed_at"]
result["message"] = f"Video generation {current_status.lower()}"
if "failureMessage" in response:
result["failure_message"] = response["failureMessage"]
invocation_data["failure_message"] = response["failureMessage"]
return result
except ClientError as e:
return {
"error": f"Failed to get invocation status: {e}",
"job_id": job_id,
"last_known_status": invocation_data.get("status", "Unknown")
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def get_prompting_guide() -> Dict[str, Any]:
"""
Get comprehensive prompting guidelines for Amazon Nova Reel video generation.
Returns:
Dict containing prompting best practices and examples
"""
return get_prompting_guidelines()
def main():
"""Main function to run the MCP server"""
parser = argparse.ArgumentParser(description="Amazon Nova Reel 1.1 MCP Server")
parser.add_argument("--aws-access-key-id", help="AWS Access Key ID")
parser.add_argument("--aws-secret-access-key", help="AWS Secret Access Key")
parser.add_argument("--aws-session-token", help="AWS Session Token (for temporary credentials)")
parser.add_argument("--aws-profile", help="AWS Profile name (alternative to explicit credentials)")
parser.add_argument("--aws-region", default="us-east-1", help="AWS Region")
parser.add_argument("--s3-bucket", help="S3 bucket name for video output")
args = parser.parse_args()
# Set global configuration from args or environment variables
global aws_access_key_id, aws_secret_access_key, aws_session_token, aws_profile, aws_region, s3_bucket
aws_access_key_id = args.aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID")
aws_secret_access_key = args.aws_secret_access_key or os.getenv("AWS_SECRET_ACCESS_KEY")
aws_session_token = args.aws_session_token or os.getenv("AWS_SESSION_TOKEN")
aws_profile = args.aws_profile or os.getenv("AWS_PROFILE")
aws_region = args.aws_region or os.getenv("AWS_REGION", "us-east-1")
s3_bucket = args.s3_bucket or os.getenv("S3_BUCKET")
# Validate configuration - need either profile OR explicit credentials + S3 bucket
if not s3_bucket:
print("Error: Missing required S3_BUCKET configuration.", file=sys.stderr)
print("Please provide --s3-bucket or S3_BUCKET env var", file=sys.stderr)
sys.exit(1)
# Check if we have valid credential configuration
has_explicit_creds = aws_access_key_id and aws_secret_access_key
has_profile = aws_profile
if not has_explicit_creds and not has_profile:
print("Error: Missing AWS credentials configuration.", file=sys.stderr)
print("Please provide either:", file=sys.stderr)
print(" Option 1: --aws-access-key-id and --aws-secret-access-key (with optional --aws-session-token)", file=sys.stderr)
print(" Option 2: --aws-profile", file=sys.stderr)
print(" Option 3: Set corresponding environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_PROFILE)", file=sys.stderr)
print(" Option 4: Configure default AWS credentials (e.g., via aws configure)", file=sys.stderr)
sys.exit(1)
if has_explicit_creds and has_profile:
print("Warning: Both explicit credentials and AWS profile provided. Using explicit credentials.", file=sys.stderr)
aws_profile = None # Clear profile to avoid confusion
# Remove s3:// prefix if present
if s3_bucket.startswith("s3://"):
s3_bucket = s3_bucket[5:]
# Load existing invocations
load_invocations()
# Initialize AWS client
try:
initialize_aws_client()
print(f"Nova Reel MCP Server initialized with region: {aws_region}, bucket: {s3_bucket}", file=sys.stderr)
print(f"Loaded {len(active_invocations)} existing invocations", file=sys.stderr)
except AWSConfigError as e:
print(f"AWS configuration error: {e}", file=sys.stderr)
sys.exit(1)
# Run MCP server
mcp.run(transport="stdio")
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/src/novareel_mcp_server/server_http.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Amazon Nova Reel 1.1 MCP Server - HTTP Streaming Version
Provides tools for video generation using AWS Bedrock Nova Reel model via HTTP Streaming transport.
"""
import argparse
import asyncio
import json
import os
import sys
import random
import time
from datetime import datetime
from typing import Optional, Dict, Any, List
import boto3
from botocore.exceptions import ClientError, NoCredentialsError
from fastmcp import FastMCP
from .prompting_guide import get_prompting_guidelines
# Create MCP server with HTTP transport
mcp = FastMCP("Amazon Nova Reel 1.1 HTTP")
# Global variables for AWS configuration
aws_access_key_id: Optional[str] = None
aws_secret_access_key: Optional[str] = None
aws_session_token: Optional[str] = None
aws_profile: Optional[str] = None
aws_region: Optional[str] = None
s3_bucket: Optional[str] = None
bedrock_client = None
# Model configuration
MODEL_ID = "amazon.nova-reel-v1:1"
SLEEP_SECONDS = 5 # Interval for checking video generation progress
# Persistent storage for tracking invocations
INVOCATIONS_FILE = os.path.expanduser("~/.novareel_invocations_http.json")
active_invocations = {}
def load_invocations():
"""Load invocations from persistent storage"""
global active_invocations
try:
if os.path.exists(INVOCATIONS_FILE):
with open(INVOCATIONS_FILE, 'r') as f:
active_invocations = json.load(f)
except Exception as e:
print(f"Warning: Could not load invocations file: {e}", file=sys.stderr)
active_invocations = {}
def save_invocations():
"""Save invocations to persistent storage"""
try:
with open(INVOCATIONS_FILE, 'w') as f:
json.dump(active_invocations, f, indent=2)
except Exception as e:
print(f"Warning: Could not save invocations file: {e}", file=sys.stderr)
class NovaReelError(Exception):
"""Base exception for Nova Reel operations"""
pass
class AWSConfigError(NovaReelError):
"""AWS configuration error"""
pass
class VideoGenerationError(NovaReelError):
"""Video generation error"""
pass
def initialize_aws_client():
"""Initialize AWS Bedrock client with provided credentials or profile"""
global bedrock_client
if not s3_bucket:
raise AWSConfigError("Missing required S3_BUCKET configuration")
try:
# Option 1: Use AWS Profile
if aws_profile:
print(f"Using AWS profile: {aws_profile}", file=sys.stderr)
session = boto3.Session(profile_name=aws_profile, region_name=aws_region)
bedrock_client = session.client("bedrock-runtime")
# Option 2: Use explicit credentials
elif aws_access_key_id and aws_secret_access_key:
print("Using explicit AWS credentials", file=sys.stderr)
client_kwargs = {
"service_name": "bedrock-runtime",
"region_name": aws_region,
"aws_access_key_id": aws_access_key_id,
"aws_secret_access_key": aws_secret_access_key
}
# Add session token if provided (for temporary credentials)
if aws_session_token:
client_kwargs["aws_session_token"] = aws_session_token
print("Using temporary credentials with session token", file=sys.stderr)
bedrock_client = boto3.client(**client_kwargs)
# Option 3: Use default credential chain
else:
print("Using default AWS credential chain", file=sys.stderr)
bedrock_client = boto3.client("bedrock-runtime", region_name=aws_region)
# Test the connection with a simple operation
# Note: bedrock-runtime doesn't have list_foundation_models, that's in bedrock client
# We'll just create the client and let the first actual call test the connection
except NoCredentialsError:
raise AWSConfigError("No valid AWS credentials found. Please provide explicit credentials, set AWS_PROFILE, or configure default credentials.")
except ClientError as e:
raise AWSConfigError(f"AWS client error: {e}")
except Exception as e:
raise AWSConfigError(f"Failed to initialize AWS client: {e}")
@mcp.tool()
async def start_async_invoke(
prompt: str,
duration_seconds: int = 12,
fps: int = 24,
dimension: str = "1280x720",
seed: Optional[int] = None,
task_type: str = "MULTI_SHOT_AUTOMATED"
) -> Dict[str, Any]:
"""
Start asynchronous video generation with Amazon Nova Reel.
Args:
prompt: Text description for video generation. See prompting guidelines for best practices.
duration_seconds: Video duration in seconds (must be multiple of 6, range 12-120)
fps: Frames per second (24 recommended)
dimension: Video dimensions (1280x720, 1920x1080, etc.)
seed: Random seed for reproducible results (optional)
task_type: Task type (MULTI_SHOT_AUTOMATED recommended)
Returns:
Dict containing invocation details and job information
"""
try:
if not bedrock_client:
initialize_aws_client()
# Validate duration
if duration_seconds < 12 or duration_seconds > 120 or duration_seconds % 6 != 0:
return {
"error": "Duration must be a multiple of 6 in range [12, 120]",
"valid_durations": [12, 18, 24, 30, 36, 42, 48, 54, 60, 66, 72, 78, 84, 90, 96, 102, 108, 114, 120]
}
# Generate seed if not provided
if seed is None:
seed = random.randint(0, 2147483648)
# Prepare model input
model_input = {
"taskType": task_type,
"multiShotAutomatedParams": {"text": prompt},
"videoGenerationConfig": {
"durationSeconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed,
},
}
# Start async invocation
invocation = bedrock_client.start_async_invoke(
modelId=MODEL_ID,
modelInput=model_input,
outputDataConfig={"s3OutputDataConfig": {"s3Uri": f"s3://{s3_bucket}"}},
)
invocation_arn = invocation["invocationArn"]
job_id = invocation_arn.split("/")[-1]
s3_location = f"s3://{s3_bucket}/{job_id}"
# Store invocation details
invocation_data = {
"invocation_arn": invocation_arn,
"job_id": job_id,
"prompt": prompt,
"duration_seconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed,
"task_type": task_type,
"s3_location": s3_location,
"status": "InProgress",
"created_at": datetime.now().isoformat(),
"video_url": None
}
active_invocations[job_id] = invocation_data
save_invocations() # Save to persistent storage
return {
"success": True,
"invocation_arn": invocation_arn,
"job_id": job_id,
"status": "InProgress",
"s3_location": s3_location,
"estimated_video_url": f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4",
"prompt": prompt,
"config": {
"duration_seconds": duration_seconds,
"fps": fps,
"dimension": dimension,
"seed": seed
},
"message": "Video generation started. Use get_async_invoke to check progress."
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except ClientError as e:
return {"error": f"AWS API error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def list_async_invokes() -> Dict[str, Any]:
"""
List all tracked async video generation invocations.
Returns:
Dict containing list of all invocations with their current status
"""
try:
if not bedrock_client:
initialize_aws_client()
# Update status for all active invocations
updated_invocations = []
for job_id, invocation_data in active_invocations.items():
try:
# Get current status from AWS
response = bedrock_client.get_async_invoke(
invocationArn=invocation_data["invocation_arn"]
)
# Update status
current_status = response["status"]
invocation_data["status"] = current_status
if current_status == "Completed":
invocation_data["video_url"] = f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4"
invocation_data["completed_at"] = datetime.now().isoformat()
elif current_status in ["Failed", "Cancelled"]:
invocation_data["failed_at"] = datetime.now().isoformat()
if "failureMessage" in response:
invocation_data["failure_message"] = response["failureMessage"]
updated_invocations.append({
"job_id": job_id,
"status": current_status,
"prompt": invocation_data["prompt"][:100] + "..." if len(invocation_data["prompt"]) > 100 else invocation_data["prompt"],
"created_at": invocation_data["created_at"],
"video_url": invocation_data.get("video_url"),
"duration_seconds": invocation_data["duration_seconds"]
})
except ClientError as e:
# If we can't get status, mark as unknown
invocation_data["status"] = "Unknown"
invocation_data["error"] = str(e)
updated_invocations.append({
"job_id": job_id,
"status": "Unknown",
"prompt": invocation_data["prompt"][:100] + "..." if len(invocation_data["prompt"]) > 100 else invocation_data["prompt"],
"created_at": invocation_data["created_at"],
"error": str(e)
})
return {
"success": True,
"total_invocations": len(updated_invocations),
"invocations": updated_invocations,
"summary": {
"in_progress": len([inv for inv in updated_invocations if inv["status"] == "InProgress"]),
"completed": len([inv for inv in updated_invocations if inv["status"] == "Completed"]),
"failed": len([inv for inv in updated_invocations if inv["status"] in ["Failed", "Cancelled"]]),
"unknown": len([inv for inv in updated_invocations if inv["status"] == "Unknown"])
}
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def get_async_invoke(identifier: str) -> Dict[str, Any]:
"""
Get detailed information about a specific async video generation invocation.
Args:
identifier: Either job_id or invocation_arn
Returns:
Dict containing detailed invocation information and video URL if completed
"""
try:
if not bedrock_client:
initialize_aws_client()
# Find invocation by job_id or invocation_arn
invocation_data = None
job_id = None
if identifier in active_invocations:
# Direct job_id lookup
job_id = identifier
invocation_data = active_invocations[identifier]
else:
# Search by invocation_arn
for jid, data in active_invocations.items():
if data["invocation_arn"] == identifier:
job_id = jid
invocation_data = data
break
if not invocation_data:
return {
"error": f"Invocation not found: {identifier}",
"suggestion": "Use list_async_invokes to see all tracked invocations"
}
# Get current status from AWS
try:
response = bedrock_client.get_async_invoke(
invocationArn=invocation_data["invocation_arn"]
)
current_status = response["status"]
invocation_data["status"] = current_status
# Prepare detailed response
result = {
"success": True,
"job_id": job_id,
"invocation_arn": invocation_data["invocation_arn"],
"status": current_status,
"prompt": invocation_data["prompt"],
"config": {
"duration_seconds": invocation_data["duration_seconds"],
"fps": invocation_data["fps"],
"dimension": invocation_data["dimension"],
"seed": invocation_data["seed"],
"task_type": invocation_data["task_type"]
},
"s3_location": invocation_data["s3_location"],
"created_at": invocation_data["created_at"]
}
if current_status == "Completed":
video_url = f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com/{job_id}/output.mp4"
invocation_data["video_url"] = video_url
invocation_data["completed_at"] = datetime.now().isoformat()
result["video_url"] = video_url
result["completed_at"] = invocation_data["completed_at"]
result["message"] = "Video generation completed successfully!"
elif current_status == "InProgress":
result["message"] = "Video generation is still in progress. Check again in a few moments."
elif current_status in ["Failed", "Cancelled"]:
invocation_data["failed_at"] = datetime.now().isoformat()
result["failed_at"] = invocation_data["failed_at"]
result["message"] = f"Video generation {current_status.lower()}"
if "failureMessage" in response:
result["failure_message"] = response["failureMessage"]
invocation_data["failure_message"] = response["failureMessage"]
return result
except ClientError as e:
return {
"error": f"Failed to get invocation status: {e}",
"job_id": job_id,
"last_known_status": invocation_data.get("status", "Unknown")
}
except AWSConfigError as e:
return {"error": f"AWS configuration error: {e}"}
except Exception as e:
return {"error": f"Unexpected error: {e}"}
@mcp.tool()
async def get_prompting_guide() -> Dict[str, Any]:
"""
Get comprehensive prompting guidelines for Amazon Nova Reel video generation.
Returns:
Dict containing prompting best practices and examples
"""
return get_prompting_guidelines()
def main():
"""Main function to run the MCP server with HTTP streaming transport"""
parser = argparse.ArgumentParser(description="Amazon Nova Reel 1.1 MCP Server - HTTP Streaming Version")
parser.add_argument("--aws-access-key-id", help="AWS Access Key ID")
parser.add_argument("--aws-secret-access-key", help="AWS Secret Access Key")
parser.add_argument("--aws-session-token", help="AWS Session Token (for temporary credentials)")
parser.add_argument("--aws-profile", help="AWS Profile name (alternative to explicit credentials)")
parser.add_argument("--aws-region", default="us-east-1", help="AWS Region")
parser.add_argument("--s3-bucket", help="S3 bucket name for video output")
parser.add_argument("--host", default="0.0.0.0", help="Host to bind to")
parser.add_argument("--port", type=int, default=8001, help="Port to bind to")
args = parser.parse_args()
# Set global configuration from args or environment variables
global aws_access_key_id, aws_secret_access_key, aws_session_token, aws_profile, aws_region, s3_bucket
aws_access_key_id = args.aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID")
aws_secret_access_key = args.aws_secret_access_key or os.getenv("AWS_SECRET_ACCESS_KEY")
aws_session_token = args.aws_session_token or os.getenv("AWS_SESSION_TOKEN")
aws_profile = args.aws_profile or os.getenv("AWS_PROFILE")
aws_region = args.aws_region or os.getenv("AWS_REGION", "us-east-1")
s3_bucket = args.s3_bucket or os.getenv("S3_BUCKET")
# Validate configuration - need either profile OR explicit credentials + S3 bucket
if not s3_bucket:
print("Error: Missing required S3_BUCKET configuration.", file=sys.stderr)
print("Please provide --s3-bucket or S3_BUCKET env var", file=sys.stderr)
sys.exit(1)
# Check if we have valid credential configuration
has_explicit_creds = aws_access_key_id and aws_secret_access_key
has_profile = aws_profile
if not has_explicit_creds and not has_profile:
print("Error: Missing AWS credentials configuration.", file=sys.stderr)
print("Please provide either:", file=sys.stderr)
print(" Option 1: --aws-access-key-id and --aws-secret-access-key (with optional --aws-session-token)", file=sys.stderr)
print(" Option 2: --aws-profile", file=sys.stderr)
print(" Option 3: Set corresponding environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_PROFILE)", file=sys.stderr)
print(" Option 4: Configure default AWS credentials (e.g., via aws configure)", file=sys.stderr)
sys.exit(1)
if has_explicit_creds and has_profile:
print("Warning: Both explicit credentials and AWS profile provided. Using explicit credentials.", file=sys.stderr)
aws_profile = None # Clear profile to avoid confusion
# Remove s3:// prefix if present
if s3_bucket.startswith("s3://"):
s3_bucket = s3_bucket[5:]
# Load existing invocations
load_invocations()
# Initialize AWS client
try:
initialize_aws_client()
print(f"Nova Reel MCP Server (HTTP Streaming) initialized with region: {aws_region}, bucket: {s3_bucket}", file=sys.stderr)
print(f"Loaded {len(active_invocations)} existing invocations", file=sys.stderr)
print(f"Starting server on {args.host}:{args.port}", file=sys.stderr)
except AWSConfigError as e:
print(f"AWS configuration error: {e}", file=sys.stderr)
sys.exit(1)
# Run MCP server with HTTP streaming transport
mcp.run(transport="http", host=args.host, port=args.port)
if __name__ == "__main__":
main()
```