# Imagen MCP Server - Implementation Plan
## Executive Summary
This document outlines the implementation plan for an MCP (Model Context Protocol) server that provides image generation capabilities using Google's Imagen model and other models supported by the Nexos.ai platform. The server will be built using Python and the FastMCP framework, following test-driven development (TDD) principles.
## Research Sources
The following sources were consulted during the planning phase:
1. **Nexos.ai Documentation**
- Gateway API: https://docs.nexos.ai/gateway-api
- Models: https://docs.nexos.ai/models
- US Models: https://docs.nexos.ai/models/nexos.ai-us-models
2. **MCP Python SDK**
- GitHub Repository: https://github.com/modelcontextprotocol/python-sdk
- Official Documentation: https://modelcontextprotocol.github.io/python-sdk/
3. **FastMCP Framework**
- Tutorial: https://gofastmcp.com/tutorials/create-mcp-server
- Testing Guide: https://gofastmcp.com/patterns/testing
4. **Model Context Protocol Specification**
- https://modelcontextprotocol.io/specification/
5. **Existing Project Research**
- Local file: `docs/imagen-mcp-session-style-research.md`
## 1. Project Overview
### 1.1 Goals
The MCP server will provide the following features:
1. **Simple Prompt-to-Image Generation**: A straightforward tool that takes a text prompt and returns a generated image.
2. **Multi-Attempt Image Generation with Background Processing**:
- First API call accepts a prompt and number of attempts
- Blocks until the first image is ready, then returns it
- Continues generating remaining images in the background
- Subsequent calls return already-generated images or block until the next one is ready
- Saves wall-clock time when user requests images slower than generation speed
3. **Model Catalog**: A resource that lists all available image generation models with their capabilities, limitations, and use case descriptions.
### 1.2 Target Platform
- **API Endpoint**: `https://api.nexos.ai/v1`
- **Authentication**: OAuth2 Bearer token
- **Primary Models**: Imagen 4, Imagen 4 Fast, Imagen 4 Ultra, FLUX 1.1 Pro, GPT Image 1, GPT Image 1 mini, Dall-E 3
## 2. Architecture
### 2.1 High-Level Architecture
```
┌─────────────────────────────────────────────────────────────────┐
│ MCP Client │
│ (Claude, Cursor, etc.) │
└─────────────────────────────────────────────────────────────────┘
│
│ MCP Protocol (stdio/HTTP)
▼
┌─────────────────────────────────────────────────────────────────┐
│ Imagen MCP Server │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ FastMCP Layer ││
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ││
│ │ │ Tools │ │ Resources │ │ Prompts │ ││
│ │ └──────────────┘ └──────────────┘ └──────────────┘ ││
│ └─────────────────────────────────────────────────────────────┘│
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ Core Services Layer ││
│ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ││
│ │ │ ImageGen │ │ Session │ │ Model │ ││
│ │ │ Service │ │ Manager │ │ Registry │ ││
│ │ └──────────────┘ └──────────────┘ └──────────────┘ ││
│ └─────────────────────────────────────────────────────────────┘│
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ Nexos.ai Client Layer ││
│ │ ┌──────────────┐ ┌──────────────┐ ││
│ │ │ HTTP Client │ │ Auth Handler │ ││
│ │ │ (httpx) │ │ │ ││
│ │ └──────────────┘ └──────────────┘ ││
│ └─────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────┘
│
│ HTTPS
▼
┌─────────────────────────────────────────────────────────────────┐
│ Nexos.ai API │
│ https://api.nexos.ai/v1 │
└─────────────────────────────────────────────────────────────────┘
```
### 2.2 Module Structure
```
Imagen_MCP/
├── __init__.py # Package exports
├── server.py # FastMCP server definition
├── tools/
│ ├── __init__.py
│ ├── generate_image.py # Simple image generation tool
│ └── batch_generate.py # Multi-attempt generation tool
├── resources/
│ ├── __init__.py
│ └── models.py # Model catalog resource
├── services/
│ ├── __init__.py
│ ├── nexos_client.py # Nexos.ai API client
│ ├── session_manager.py # Background generation session manager
│ └── model_registry.py # Model information registry
├── models/
│ ├── __init__.py
│ ├── image.py # Image data models
│ ├── generation.py # Generation request/response models
│ └── session.py # Session state models
└── config.py # Configuration management
```
## 3. Feature Specifications
### 3.1 Simple Image Generation Tool
**Tool Name**: `generate_image`
**Description**: Generate a single image from a text prompt using a specified model.
**Input Schema**:
```python
class GenerateImageInput(BaseModel):
prompt: str = Field(description="Text description of the image to generate")
model: str = Field(
default="imagen-4",
description="Model to use for generation"
)
size: str = Field(
default="1024x1024",
description="Image size (256x256, 512x512, 1024x1024, 1792x1024, 1024x1792)"
)
quality: str = Field(
default="standard",
description="Image quality (standard, hd)"
)
style: str = Field(
default="vivid",
description="Image style (vivid, natural)"
)
```
**Output**: Base64-encoded image data with metadata.
### 3.2 Multi-Attempt Image Generation Tool
**Tool Name**: `start_image_batch`
**Description**: Start generating multiple image variations from a prompt. Returns the first image immediately and continues generating in the background.
**Input Schema**:
```python
class StartImageBatchInput(BaseModel):
prompt: str = Field(description="Text description of the image to generate")
count: int = Field(
default=4,
ge=2,
le=10,
description="Number of image variations to generate"
)
model: str = Field(default="imagen-4")
size: str = Field(default="1024x1024")
quality: str = Field(default="standard")
style: str = Field(default="vivid")
```
**Output**: Session ID and the first generated image.
---
**Tool Name**: `get_next_image`
**Description**: Retrieve the next available image from a batch generation session. Blocks until an image is ready if none are available.
**Input Schema**:
```python
class GetNextImageInput(BaseModel):
session_id: str = Field(description="Session ID from start_image_batch")
timeout: float = Field(
default=60.0,
description="Maximum seconds to wait for next image"
)
```
**Output**: Next available image or status indicating completion/timeout.
---
**Tool Name**: `get_batch_status`
**Description**: Get the current status of a batch generation session.
**Input Schema**:
```python
class GetBatchStatusInput(BaseModel):
session_id: str = Field(description="Session ID from start_image_batch")
```
**Output**: Session status including completed count, pending count, and any errors.
### 3.3 Model Catalog Resource
**Resource URI**: `models://image-generation`
**Description**: Returns a comprehensive list of all available image generation models with their capabilities.
**Output Schema**:
```python
class ModelInfo(BaseModel):
id: str
name: str
provider: str
description: str
capabilities: ModelCapabilities
rate_limit: RateLimitInfo
use_cases: list[str]
strengths: list[str]
weaknesses: list[str]
class ModelCapabilities(BaseModel):
max_images_per_request: int
supported_sizes: list[str]
supports_hd_quality: bool
supports_style_parameter: bool
max_prompt_length: int
class RateLimitInfo(BaseModel):
messages_per_period: int
period_hours: int
category: str
```
## 4. Session Management Design
### 4.1 Session State Machine
```
┌─────────────┐
│ CREATED │
└──────┬──────┘
│ start_generation()
▼
┌─────────────┐
│ GENERATING │◄────────────────┐
└──────┬──────┘ │
│ │
├── image_ready() ───────┤
│ │
▼ │
┌─────────────┐ │
│ PARTIAL │─────────────────┘
└──────┬──────┘ (more images pending)
│
│ all_images_complete()
▼
┌─────────────┐
│ COMPLETED │
└─────────────┘
│ error() or timeout()
▼
┌─────────────┐
│ FAILED │
└─────────────┘
```
### 4.2 Session Data Structure
```python
@dataclass
class GenerationSession:
id: str
prompt: str
model: str
requested_count: int
created_at: datetime
status: SessionStatus
# Image storage
completed_images: list[GeneratedImage]
pending_count: int
# Tracking
next_image_index: int # For get_next_image
errors: list[GenerationError]
# Background task reference
generation_task: asyncio.Task | None
```
### 4.3 Background Generation Flow
```python
async def background_generation(session: GenerationSession):
"""Generate images in background, making them available as they complete."""
for i in range(session.requested_count):
try:
# Generate single image
image = await nexos_client.generate_image(
prompt=session.prompt,
model=session.model,
# ... other params
)
# Add to completed images (thread-safe)
async with session.lock:
session.completed_images.append(image)
session.pending_count -= 1
# Notify any waiting consumers
session.image_ready_event.set()
except Exception as e:
session.errors.append(GenerationError(index=i, error=str(e)))
session.status = SessionStatus.COMPLETED
```
## 5. Test-Driven Development Plan
### 5.1 Test Categories
The test suite will be organized into the following categories:
1. **Unit Tests**: Test individual components in isolation
2. **Integration Tests**: Test component interactions
3. **API Mock Tests**: Test Nexos.ai API interactions with mocked responses
4. **End-to-End Tests**: Test complete workflows
### 5.2 Test File Structure
```
tests/
├── conftest.py # Shared fixtures
├── unit/
│ ├── test_nexos_client.py # Nexos.ai client unit tests
│ ├── test_session_manager.py # Session manager unit tests
│ ├── test_model_registry.py # Model registry unit tests
│ └── test_models.py # Data model validation tests
├── integration/
│ ├── test_generate_image.py # Simple generation integration
│ ├── test_batch_generation.py # Batch generation integration
│ └── test_model_catalog.py # Model catalog integration
├── mocks/
│ ├── __init__.py
│ └── nexos_api.py # Nexos.ai API mock responses
└── e2e/
└── test_mcp_client.py # Full MCP client tests
```
### 5.3 Detailed Test Plan
#### 5.3.1 Unit Tests - Nexos.ai Client (`test_nexos_client.py`)
```python
# Test: Client initialization with API key
def test_client_initialization_with_api_key():
"""Client should initialize with valid API key from environment."""
# Test: Client initialization without API key raises error
def test_client_initialization_without_api_key_raises():
"""Client should raise ConfigurationError when API key is missing."""
# Test: Generate image request formatting
async def test_generate_image_request_format():
"""Generate image should send correctly formatted request to API."""
# Test: Generate image response parsing
async def test_generate_image_response_parsing():
"""Client should correctly parse API response into GeneratedImage."""
# Test: Generate image with all parameters
async def test_generate_image_with_all_parameters():
"""All optional parameters should be included in API request."""
# Test: API error handling - 400 Bad Request
async def test_api_error_400_handling():
"""Client should raise InvalidRequestError on 400 response."""
# Test: API error handling - 401 Unauthorized
async def test_api_error_401_handling():
"""Client should raise AuthenticationError on 401 response."""
# Test: API error handling - 429 Rate Limited
async def test_api_error_429_handling():
"""Client should raise RateLimitError on 429 response."""
# Test: API error handling - 500 Server Error
async def test_api_error_500_handling():
"""Client should raise ServerError on 500 response."""
# Test: Retry logic on transient errors
async def test_retry_on_transient_errors():
"""Client should retry on 5xx errors with exponential backoff."""
# Test: List models request
async def test_list_models():
"""Client should correctly fetch and parse available models."""
```
#### 5.3.2 Unit Tests - Session Manager (`test_session_manager.py`)
```python
# Test: Create new session
def test_create_session():
"""Session manager should create session with unique ID."""
# Test: Get session by ID
def test_get_session_by_id():
"""Session manager should retrieve session by ID."""
# Test: Get non-existent session raises error
def test_get_nonexistent_session_raises():
"""Session manager should raise SessionNotFoundError for invalid ID."""
# Test: Session expiration
async def test_session_expiration():
"""Expired sessions should be cleaned up automatically."""
# Test: Concurrent session limit
def test_concurrent_session_limit():
"""Session manager should enforce maximum concurrent sessions."""
# Test: Add image to session
async def test_add_image_to_session():
"""Adding image should update session state correctly."""
# Test: Get next image - image available
async def test_get_next_image_available():
"""Should return immediately when image is available."""
# Test: Get next image - wait for image
async def test_get_next_image_waits():
"""Should block until image becomes available."""
# Test: Get next image - timeout
async def test_get_next_image_timeout():
"""Should raise TimeoutError after specified timeout."""
# Test: Get next image - session completed
async def test_get_next_image_session_completed():
"""Should raise SessionCompletedError when no more images."""
# Test: Session status tracking
def test_session_status_transitions():
"""Session status should transition correctly through states."""
```
#### 5.3.3 Unit Tests - Model Registry (`test_model_registry.py`)
```python
# Test: Load model definitions
def test_load_model_definitions():
"""Registry should load all model definitions on initialization."""
# Test: Get model by ID
def test_get_model_by_id():
"""Registry should return correct model info by ID."""
# Test: Get all models
def test_get_all_models():
"""Registry should return list of all available models."""
# Test: Filter models by capability
def test_filter_models_by_capability():
"""Registry should filter models by specific capabilities."""
# Test: Model validation
def test_model_info_validation():
"""Model info should validate against schema."""
```
#### 5.3.4 Integration Tests - Generate Image (`test_generate_image.py`)
```python
# Test: Simple image generation tool
async def test_generate_image_tool(mcp_client):
"""generate_image tool should return valid image data."""
# Test: Generate image with custom size
async def test_generate_image_custom_size(mcp_client):
"""generate_image should respect size parameter."""
# Test: Generate image with HD quality
async def test_generate_image_hd_quality(mcp_client):
"""generate_image should request HD quality when specified."""
# Test: Generate image with different models
@pytest.mark.parametrize("model", ["imagen-4", "imagen-4-fast", "flux-1.1-pro"])
async def test_generate_image_different_models(mcp_client, model):
"""generate_image should work with all supported models."""
# Test: Generate image error handling
async def test_generate_image_invalid_prompt(mcp_client):
"""generate_image should handle invalid prompts gracefully."""
# Test: Generate image rate limiting
async def test_generate_image_rate_limit_handling(mcp_client):
"""generate_image should handle rate limits appropriately."""
```
#### 5.3.5 Integration Tests - Batch Generation (`test_batch_generation.py`)
```python
# Test: Start batch generation
async def test_start_batch_generation(mcp_client):
"""start_image_batch should return session ID and first image."""
# Test: Get next image from batch
async def test_get_next_image_from_batch(mcp_client):
"""get_next_image should return subsequent images."""
# Test: Get batch status
async def test_get_batch_status(mcp_client):
"""get_batch_status should return accurate session state."""
# Test: Complete batch workflow
async def test_complete_batch_workflow(mcp_client):
"""Full batch generation workflow should complete successfully."""
# Test: Batch generation with errors
async def test_batch_generation_partial_failure(mcp_client):
"""Batch should continue despite individual image failures."""
# Test: Multiple concurrent batches
async def test_multiple_concurrent_batches(mcp_client):
"""Multiple batch sessions should run concurrently."""
# Test: Batch timeout handling
async def test_batch_timeout_handling(mcp_client):
"""get_next_image should timeout appropriately."""
# Test: Session cleanup after completion
async def test_session_cleanup_after_completion(mcp_client):
"""Completed sessions should be cleaned up after TTL."""
```
#### 5.3.6 Integration Tests - Model Catalog (`test_model_catalog.py`)
```python
# Test: Read model catalog resource
async def test_read_model_catalog(mcp_client):
"""models://image-generation resource should return model list."""
# Test: Model catalog content validation
async def test_model_catalog_content(mcp_client):
"""Model catalog should contain expected models with valid data."""
# Test: Model capabilities accuracy
async def test_model_capabilities_accuracy(mcp_client):
"""Model capabilities should match actual API capabilities."""
```
#### 5.3.7 End-to-End Tests (`test_mcp_client.py`)
```python
# Test: Full MCP client connection
async def test_mcp_client_connection():
"""MCP client should connect to server successfully."""
# Test: List available tools
async def test_list_tools():
"""Server should expose all expected tools."""
# Test: List available resources
async def test_list_resources():
"""Server should expose model catalog resource."""
# Test: Tool execution workflow
async def test_tool_execution_workflow():
"""Complete tool execution should work end-to-end."""
# Test: Resource reading workflow
async def test_resource_reading_workflow():
"""Complete resource reading should work end-to-end."""
```
### 5.4 Test Fixtures
```python
# conftest.py
import pytest
from fastmcp.client import Client
from fastmcp.client.transports import FastMCPTransport
from Imagen_MCP.server import mcp
@pytest.fixture
async def mcp_client():
"""Provide a connected MCP client for testing."""
async with Client(transport=mcp) as client:
yield client
@pytest.fixture
def mock_nexos_api(httpx_mock):
"""Provide mocked Nexos.ai API responses."""
# Configure mock responses
httpx_mock.add_response(
url="https://api.nexos.ai/v1/images/generations",
json={
"created": 1234567890,
"data": [
{
"b64_json": "base64_encoded_image_data",
"revised_prompt": "A beautiful sunset..."
}
]
}
)
return httpx_mock
@pytest.fixture
def sample_generation_request():
"""Provide a sample generation request for testing."""
return {
"prompt": "A serene mountain landscape at sunset",
"model": "imagen-4",
"size": "1024x1024",
"quality": "standard",
"style": "vivid"
}
```
### 5.5 Mock API Responses
```python
# tests/mocks/nexos_api.py
MOCK_IMAGE_GENERATION_SUCCESS = {
"created": 1734800000,
"data": [
{
"b64_json": "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==",
"revised_prompt": "A serene mountain landscape at sunset with golden light"
}
]
}
MOCK_IMAGE_GENERATION_RATE_LIMITED = {
"error": {
"message": "Rate limit exceeded",
"type": "rate_limit_error",
"code": "rate_limit_exceeded"
}
}
MOCK_MODELS_LIST = {
"object": "list",
"data": [
{
"id": "imagen-4",
"nexos_model_id": "uuid-imagen-4",
"object": "model",
"owned_by": "google"
},
{
"id": "imagen-4-fast",
"nexos_model_id": "uuid-imagen-4-fast",
"object": "model",
"owned_by": "google"
}
]
}
```
## 6. Implementation Phases
### Phase 1: Foundation (Week 1)
1. **Setup Project Structure**
- Create module structure
- Configure dependencies in `pyproject.toml`
- Setup test infrastructure
2. **Implement Core Models**
- Define Pydantic models for requests/responses
- Implement data validation
3. **Implement Nexos.ai Client**
- HTTP client with authentication
- Image generation endpoint
- Error handling and retries
4. **Write Unit Tests for Phase 1**
- All tests in `test_nexos_client.py`
- All tests in `test_models.py`
### Phase 2: Simple Generation (Week 2)
1. **Implement Model Registry**
- Load model definitions
- Provide model information
2. **Implement Simple Generation Tool**
- `generate_image` tool
- Input validation
- Response formatting
3. **Implement Model Catalog Resource**
- `models://image-generation` resource
- Model information formatting
4. **Write Tests for Phase 2**
- All tests in `test_model_registry.py`
- All tests in `test_generate_image.py`
- All tests in `test_model_catalog.py`
### Phase 3: Batch Generation (Week 3)
1. **Implement Session Manager**
- Session creation and tracking
- Background task management
- Session cleanup
2. **Implement Batch Generation Tools**
- `start_image_batch` tool
- `get_next_image` tool
- `get_batch_status` tool
3. **Write Tests for Phase 3**
- All tests in `test_session_manager.py`
- All tests in `test_batch_generation.py`
### Phase 4: Integration & Polish (Week 4)
1. **End-to-End Testing**
- Full workflow tests
- Error scenario tests
2. **Documentation**
- API documentation
- Usage examples
- Deployment guide
3. **Performance Optimization**
- Connection pooling
- Caching where appropriate
## 7. Dependencies
### 7.1 Runtime Dependencies
```toml
[project.dependencies]
fastmcp = "^2.0.0" # MCP server framework (or mcp[cli])
httpx = "^0.27.0" # Async HTTP client
pydantic = "^2.0.0" # Data validation
```
### 7.2 Development Dependencies
```toml
[tool.poetry.group.dev.dependencies]
pytest = "^8.3.0"
pytest-asyncio = "^0.24.0"
pytest-httpx = "^0.30.0" # HTTP mocking
coverage = "^7.6.0"
inline-snapshot = "^0.13.0" # Snapshot testing
```
## 8. Configuration
### 8.1 Environment Variables
```bash
# Required
NEXOS_API_KEY=your_api_key_here
# Optional
NEXOS_API_BASE_URL=https://api.nexos.ai/v1
IMAGEN_DEFAULT_MODEL=imagen-4
IMAGEN_SESSION_TTL=3600
IMAGEN_MAX_CONCURRENT_SESSIONS=10
IMAGEN_LOG_LEVEL=INFO
```
### 8.2 Configuration Model
```python
class ImagenMCPConfig(BaseSettings):
"""Configuration for Imagen MCP Server."""
# API Configuration
nexos_api_key: str = Field(..., env="NEXOS_API_KEY")
nexos_api_base_url: str = Field(
default="https://api.nexos.ai/v1",
env="NEXOS_API_BASE_URL"
)
# Default Generation Settings
default_model: str = Field(default="imagen-4", env="IMAGEN_DEFAULT_MODEL")
# Session Management
session_ttl: int = Field(default=3600, env="IMAGEN_SESSION_TTL")
max_concurrent_sessions: int = Field(
default=10,
env="IMAGEN_MAX_CONCURRENT_SESSIONS"
)
# Logging
log_level: str = Field(default="INFO", env="IMAGEN_LOG_LEVEL")
class Config:
env_file = ".env"
```
## 9. Model Catalog Data
### 9.1 Available Models
| Model | Provider | Category | Rate Limit | Best For |
|-------|----------|----------|------------|----------|
| Imagen 4 | Google | 3 | 100/3h | General purpose, high quality |
| Imagen 4 Fast | Google | 3 | 100/3h | Quick iterations, drafts |
| Imagen 4 Ultra | Google | 3 | 100/3h | Maximum quality, final outputs |
| FLUX 1.1 Pro | Black Forest Labs | 3 | 100/3h | Artistic, creative styles |
| GPT Image 1 | OpenAI | 3 | 100/3h | Photorealistic, detailed |
| GPT Image 1 mini | OpenAI | 3 | 100/3h | Fast, cost-effective |
| Dall-E 3 | OpenAI | 3 (US only) | 100/3h | Creative, artistic |
### 9.2 Model Capabilities
```python
MODEL_CAPABILITIES = {
"imagen-4": {
"max_images_per_request": 10,
"supported_sizes": ["256x256", "512x512", "1024x1024", "1792x1024", "1024x1792"],
"supports_hd_quality": True,
"supports_style_parameter": True,
"max_prompt_length": 4096,
},
"imagen-4-fast": {
"max_images_per_request": 10,
"supported_sizes": ["256x256", "512x512", "1024x1024"],
"supports_hd_quality": False,
"supports_style_parameter": True,
"max_prompt_length": 4096,
},
# ... other models
}
```
## 10. Error Handling Strategy
### 10.1 Error Types
```python
class ImagenMCPError(Exception):
"""Base exception for Imagen MCP errors."""
class ConfigurationError(ImagenMCPError):
"""Configuration-related errors."""
class AuthenticationError(ImagenMCPError):
"""API authentication failures."""
class RateLimitError(ImagenMCPError):
"""Rate limit exceeded."""
class InvalidRequestError(ImagenMCPError):
"""Invalid request parameters."""
class GenerationError(ImagenMCPError):
"""Image generation failures."""
class SessionError(ImagenMCPError):
"""Session management errors."""
class SessionNotFoundError(SessionError):
"""Session not found."""
class SessionExpiredError(SessionError):
"""Session has expired."""
```
### 10.2 Error Response Format
```python
class ErrorResponse(BaseModel):
error: str
error_type: str
details: dict | None = None
retry_after: int | None = None # For rate limits
```
## 11. Logging Strategy
### 11.1 Log Levels
- **DEBUG**: Detailed request/response data, session state changes
- **INFO**: Tool invocations, successful generations, session lifecycle
- **WARNING**: Rate limit approaches, retries, partial failures
- **ERROR**: Generation failures, API errors, session errors
### 11.2 Structured Logging
```python
import structlog
logger = structlog.get_logger()
# Example usage
logger.info(
"image_generated",
session_id=session.id,
model=request.model,
size=request.size,
duration_ms=duration,
)
```
## 12. Future Enhancements
### 12.1 Potential Features
1. **Image-to-Image Generation**: When Nexos.ai adds support
2. **Style Consistency Sessions**: Using prompt engineering techniques
3. **Image Caching**: Cache generated images for repeated prompts
4. **Cost Tracking**: Track API usage and estimated costs
5. **Webhook Notifications**: Notify when batch generation completes
### 12.2 Scalability Considerations
1. **Horizontal Scaling**: Stateless design for multi-instance deployment
2. **External Session Store**: Redis/PostgreSQL for session persistence
3. **Queue-Based Generation**: Background job queue for large batches
## 13. Appendix
### 13.1 API Reference - Nexos.ai Image Generation
**Endpoint**: `POST /v1/images/generations`
**Request**:
```json
{
"prompt": "string (required)",
"model": "string (required, model UUID)",
"n": "integer (optional, 1-10, default 1)",
"quality": "string (optional, 'standard' or 'hd')",
"response_format": "string (optional, 'url' or 'b64_json')",
"size": "string (optional)",
"style": "string (optional, 'vivid' or 'natural')"
}
```
**Response**:
```json
{
"created": 1234567890,
"data": [
{
"b64_json": "base64_encoded_image",
"revised_prompt": "enhanced prompt used",
"url": "image_url (if response_format is 'url')"
}
]
}
```
### 13.2 MCP Tool Schema Examples
```json
{
"name": "generate_image",
"description": "Generate an image from a text prompt",
"inputSchema": {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": "Text description of the image to generate"
},
"model": {
"type": "string",
"default": "imagen-4",
"enum": ["imagen-4", "imagen-4-fast", "imagen-4-ultra", "flux-1.1-pro"]
},
"size": {
"type": "string",
"default": "1024x1024",
"enum": ["256x256", "512x512", "1024x1024", "1792x1024", "1024x1792"]
}
},
"required": ["prompt"]
}
}
