Initial commit: A1D MCP Server with Gradio interface

- Complete MCP server implementation with Gradio web interface
- Support for 6 AI tools: background removal, image upscaling, video upscaling, image vectorization, image extension, and image generation
- Dual API key support: header-based (recommended) and environment variable
- Real-time SSE processing with detailed logging
- Media preview support for images and videos
- Production-ready with comprehensive error handling
- Full documentation with MCP client configuration examples

.env.example

@@ -0,0 +1,11 @@
+# A1D MCP Server Environment Variables
+# Copy this file to .env and fill in your values
+
+# A1D API Key (Required)
+# Get your API key from: https://a1d.ai/home/api
+A1D_API_KEY=z4Zl7OhAM5r-DxyVFDN1A
+
+# Optional: Gradio Configuration
+# GRADIO_SERVER_NAME=0.0.0.0
+# GRADIO_SERVER_PORT=7860
+# GRADIO_SHARE=false

.gitignore

@@ -0,0 +1,62 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# Virtual Environment
+venv/
+env/
+ENV/
+env.bak/
+venv.bak/
+
+# Environment Variables
+.env
+.env.local
+.env.production
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db
+
+# Gradio
+gradio_cached_examples/
+flagged/
+
+# Logs
+*.log
+logs/
+
+# Temporary files
+*.tmp
+*.temp

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 A1D Team
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -0,0 +1,277 @@
+---
+title: A1d Mcp Server
+emoji: 🤖
+colorFrom: blue
+colorTo: purple
+sdk: gradio
+sdk_version: 5.17.0
+app_file: app.py
+pinned: false
+license: mit
+tags:
+  - mcp-server-track
+---
+
+# A1D MCP Server - Universal AI Tools
+
+A powerful MCP (Model Context Protocol) server built with Gradio that provides AI image and video processing tools for any MCP-compatible client. This server implements the same tools as the original [A1D MCP Server](https://github.com/AIGC-Hackers/mcp-server) but uses Gradio for easy deployment and hosting.
+
+## 🎥 Demo
+
+**MCP Server in Action**: [View Demo Recording](https://huggingface.co/spaces/aigchacker/a1d-mcp-server)
+
+*The demo shows the A1D MCP Server working with Claude Desktop and other MCP clients, featuring real-time AI image processing, seamless integration, and live preview of results.*
+
+## 🤖 Available AI Tools
+
+| Tool | Description | Use Cases |
+|------|-------------|-----------|
+| **remove_bg** | AI background removal | Remove backgrounds from photos, product images |
+| **image_upscaler** | AI image enhancement | Upscale images 2x, 4x, 8x, 16x resolution |
+| **video_upscaler** | AI video enhancement | Improve video quality and resolution |
+| **image_vectorization** | Convert to vectors | Turn images into scalable SVG graphics |
+| **image_extends** | Smart image extension | Expand image boundaries intelligently |
+| **image_generator** | Text-to-image AI | Generate images from text descriptions |
+
+## 🚀 Quick Setup
+
+### 1. Get Your API Key
+- Visit [A1D.ai](https://a1d.ai/home/api) to get your free API key
+- Optional: [Purchase credits](https://a1d.ai/pricing) for extended usage
+
+### 2. Set Environment Variable
+```bash
+export A1D_API_KEY=your_api_key_here
+```
+
+### 3. Install Dependencies
+```bash
+pip install -r requirements.txt
+```
+
+### 4. Run the Server
+
+**Option 1: Using the startup script (Recommended)**
+```bash
+python start_server.py
+```
+
+**Option 2: Direct execution**
+```bash
+python app.py
+```
+
+**Option 3: Using virtual environment**
+```bash
+source venv/bin/activate  # On Windows: venv\Scripts\activate
+python start_server.py
+```
+
+The server will start on `http://localhost:7860` with MCP server enabled.
+
+### 5. Test the Installation
+```bash
+python test_app.py
+```
+
+This will run a comprehensive test suite to verify everything is working correctly.
+
+## 🔧 MCP Server Features
+
+This Gradio app serves as both a **web interface** and a **full MCP server**, providing:
+
+### ✅ **Dual Interface Support**
+- **Web UI**: Interactive Gradio interface at `http://localhost:7860`
+- **MCP Server**: Protocol-compliant server at `http://localhost:7860/gradio_api/mcp/sse`
+
+### ✅ **Real-time Processing**
+- **Async Task Handling**: Uses SSE for real-time status updates
+- **Live Preview**: Immediate media preview in both web and MCP clients
+- **Progress Tracking**: Detailed logs of API calls and processing status
+
+### ✅ **Production Ready**
+- **Error Handling**: Comprehensive error management and user feedback
+- **API Integration**: Direct integration with A1D.ai services
+- **Scalable**: Can handle multiple concurrent requests
+
+## 🔧 MCP Client Configuration
+
+### Method 1: With API Key in Headers (Recommended)
+
+Add this to your Claude Desktop configuration file:
+
+```json
+{
+  "mcpServers": {
+    "a1d": {
+      "command": "npx",
+      "args": [
+        "mcp-remote@latest",
+        "http://localhost:7860/gradio_api/mcp/sse",
+        "--header",
+        "API_KEY:${MCP_API_KEY}"
+      ],
+      "env": {
+        "MCP_API_KEY": "your_a1d_api_key_here"
+      }
+    }
+  }
+}
+```
+
+### Method 2: With Environment Variable (Fallback)
+
+```json
+{
+  "mcpServers": {
+    "a1d-gradio": {
+      "command": "npx",
+      "args": [
+        "mcp-remote",
+        "http://localhost:7860/gradio_api/mcp/sse"
+      ]
+    }
+  }
+}
+```
+
+*Note: Requires `A1D_API_KEY` environment variable to be set.*
+
+### For Other MCP Clients
+- **Server URL:** `http://localhost:7860/gradio_api/mcp/sse`
+- **Transport Type:** SSE (Server-Sent Events)
+- **Protocol:** MCP 1.0 compatible
+- **Authentication:** API key via `API_KEY` header or environment variable
+
+### Configuration File Locations
+
+**Windows:** `%APPDATA%\Claude\claude_desktop_config.json`
+**macOS:** `~/Library/Application Support/Claude/claude_desktop_config.json`
+**Linux:** `~/.config/Claude/claude_desktop_config.json`
+
+## 🌐 Hosted Version
+
+This server is also available as a hosted Hugging Face Space:
+- **Space URL:** [https://huggingface.co/spaces/aigchacker/a1d-mcp-server](https://huggingface.co/spaces/aigchacker/a1d-mcp-server)
+- **MCP Endpoint:** `https://aigchacker-a1d-mcp-server.hf.space/gradio_api/mcp/sse`
+
+### 🔧 Environment Variables for Hugging Face Space
+
+To deploy this on your own Hugging Face Space:
+
+1. **Fork or duplicate this Space**
+2. **Go to Settings → Variables and secrets**
+3. **Add the following environment variable:**
+   - **Name**: `A1D_API_KEY`
+   - **Value**: Your A1D API key from [A1D.ai](https://a1d.ai/home/api)
+   - **Type**: Secret (recommended for security)
+4. **Save and restart the Space**
+
+### 🔗 MCP Client Configuration for Hosted Version
+
+**Method 1: With API Key in Headers (Recommended)**
+
+```json
+{
+  "mcpServers": {
+    "a1d-hosted": {
+      "command": "npx",
+      "args": [
+        "mcp-remote@latest",
+        "https://aigchacker-a1d-mcp-server.hf.space/gradio_api/mcp/sse",
+        "--header",
+        "API_KEY:${MCP_API_KEY}"
+      ],
+      "env": {
+        "MCP_API_KEY": "your_a1d_api_key_here"
+      }
+    }
+  }
+}
+```
+
+**Method 2: Environment Variable (if Space has API key configured)**
+
+```json
+{
+  "mcpServers": {
+    "a1d-hosted": {
+      "command": "npx",
+      "args": [
+        "mcp-remote",
+        "https://aigchacker-a1d-mcp-server.hf.space/gradio_api/mcp/sse"
+      ]
+    }
+  }
+}
+```
+
+## 💡 How to Use
+
+Once configured, simply ask your AI assistant to help with image or video tasks:
+
+- *"Remove the background from this image: https://example.com/photo.jpg"*
+- *"Upscale this image to 4x resolution: https://example.com/image.png"*
+- *"Convert this photo to a vector graphic: https://example.com/logo.jpg"*
+- *"Generate an image of a sunset over mountains"*
+
+## 🛠️ Development
+
+### Local Development
+```bash
+git clone https://github.com/your-repo/a1d-mcp-server-hf.git
+cd a1d-mcp-server-hf
+pip install -r requirements.txt
+export A1D_API_KEY=your_api_key_here
+python app.py
+```
+
+### Project Structure
+```
+a1d-mcp-server-hf/
+├── app.py              # Main Gradio application
+├── config.py           # Configuration settings
+├── utils.py            # Utility functions and API client
+├── mcp_handler.py      # MCP request handler with header-based API keys
+├── start_server.py     # Server startup script
+├── test_app.py         # Test suite
+├── requirements.txt    # Python dependencies
+├── .env.example        # Environment variables example
+├── .gitignore          # Git ignore file
+├── LICENSE             # MIT license
+└── README.md          # This file
+```
+
+## 🔐 Security
+
+- **User-provided credentials**: This server supports multiple API key methods:
+  - **Header-based**: API keys passed via MCP client headers (recommended)
+  - **Environment variables**: Server-side API key configuration
+- **No stored secrets**: All API keys are handled per-request, nothing is stored server-side
+- **Multi-user support**: Each user can use their own API key via headers
+- **HTTPS recommended**: Use HTTPS in production environments
+
+## 🤝 Contributing
+
+We welcome contributions! Please:
+
+1. Fork the repository
+2. Create a feature branch
+3. Follow the existing code style
+4. Add tests for new features
+5. Submit a pull request
+
+## 📄 License
+
+This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
+
+## 🔗 Related Projects
+
+- **Original MCP Server:** [AIGC-Hackers/mcp-server](https://github.com/AIGC-Hackers/mcp-server)
+- **A1D API Documentation:** [A1D.ai API Docs](https://a1d.ai/api/quick-start)
+- **Gradio MCP Guide:** [Building MCP Server with Gradio](https://www.gradio.app/guides/building-mcp-server-with-gradio)
+
+---
+
+**Built with ❤️ by the A1D Team**  
+[A1D.ai](https://a1d.ai) • [GitHub](https://github.com/AIGC-Hackers) • [API Docs](https://a1d.ai/api/quick-start)

app.py

@@ -0,0 +1,448 @@
+"""
+A1D MCP Server - Gradio Application
+Universal AI Tools for image and video processing
+"""
+
+import gradio as gr
+import os
+from typing import Optional, Tuple, Union
+from utils import A1DAPIClient, validate_url, validate_scale, prepare_request_data, format_response_with_preview
+from config import GRADIO_CONFIG, TOOLS_CONFIG
+from mcp_handler import get_api_key_from_headers
+
+
+# Initialize API client
+def get_api_client():
+    """Get API client with current API key"""
+    # Try to get API key from multiple sources
+    api_key = None
+
+    # 1. Try from request headers (for MCP clients)
+    try:
+        request = gr.request()
+        if request and hasattr(request, 'headers'):
+            api_key = get_api_key_from_headers(dict(request.headers))
+    except:
+        pass
+
+    # 2. Fallback to environment variable
+    if not api_key:
+        api_key = os.getenv("A1D_API_KEY")
+
+    if not api_key:
+        raise ValueError(
+            "API key is required. Set A1D_API_KEY environment variable or provide API_KEY in request headers.")
+
+    return A1DAPIClient(api_key)
+
+
+def remove_bg(image_url: str) -> Tuple[str, Optional[str]]:
+    """Remove background from images using AI.
+
+    Args:
+        image_url: The URL of the image to remove background from
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    try:
+        if not validate_url(image_url):
+            return "❌ Error: Invalid image URL format", None
+
+        client = get_api_client()
+        data = prepare_request_data("remove_bg", image_url=image_url)
+
+        # Use the new method that waits for result
+        response = client.make_request_with_result(
+            TOOLS_CONFIG["remove_bg"]["api_endpoint"],
+            data,
+            timeout=120  # 2 minutes timeout
+        )
+
+        return format_response_with_preview(response, "remove_bg")
+
+    except Exception as e:
+        return f"❌ Error: {str(e)}", None
+
+
+def image_upscaler(image_url: str, scale: int = 2) -> Tuple[str, Optional[str]]:
+    """Upscale images using AI with specified scale factor.
+
+    Args:
+        image_url: The URL of the image to upscale
+        scale: Scale factor for upscaling (2, 4, 8, or 16). Default: 2
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    try:
+        if not validate_url(image_url):
+            return "❌ Error: Invalid image URL format", None
+
+        if not validate_scale(scale):
+            return "❌ Error: Scale must be 2, 4, 8, or 16", None
+
+        client = get_api_client()
+        data = prepare_request_data(
+            "image_upscaler", image_url=image_url, scale=scale)
+
+        response = client.make_request_with_result(
+            TOOLS_CONFIG["image_upscaler"]["api_endpoint"],
+            data,
+            timeout=120
+        )
+
+        return format_response_with_preview(response, "image_upscaler")
+
+    except Exception as e:
+        return f"❌ Error: {str(e)}", None
+
+
+def video_upscaler(video_url: str) -> Tuple[str, Optional[str]]:
+    """Upscale videos using AI.
+
+    Args:
+        video_url: The URL of the video to upscale
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    try:
+        if not validate_url(video_url):
+            return "❌ Error: Invalid video URL format", None
+
+        client = get_api_client()
+        data = prepare_request_data("video_upscaler", video_url=video_url)
+
+        response = client.make_request_with_result(
+            TOOLS_CONFIG["video_upscaler"]["api_endpoint"],
+            data,
+            timeout=300  # 5 minutes for video processing
+        )
+
+        return format_response_with_preview(response, "video_upscaler")
+
+    except Exception as e:
+        return f"❌ Error: {str(e)}", None
+
+
+def image_vectorization(image_url: str) -> Tuple[str, Optional[str]]:
+    """Convert images to vector format using AI.
+
+    Args:
+        image_url: The URL of the image to vectorize
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    try:
+        if not validate_url(image_url):
+            return "❌ Error: Invalid image URL format", None
+
+        client = get_api_client()
+        data = prepare_request_data("image_vectorization", image_url=image_url)
+
+        response = client.make_request_with_result(
+            TOOLS_CONFIG["image_vectorization"]["api_endpoint"],
+            data,
+            timeout=120
+        )
+
+        return format_response_with_preview(response, "image_vectorization")
+
+    except Exception as e:
+        return f"❌ Error: {str(e)}", None
+
+
+def image_extends(image_url: str) -> Tuple[str, Optional[str]]:
+    """Extend images using AI.
+
+    Args:
+        image_url: The URL of the image to extend
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    try:
+        if not validate_url(image_url):
+            return "❌ Error: Invalid image URL format", None
+
+        client = get_api_client()
+        data = prepare_request_data("image_extends", image_url=image_url)
+
+        response = client.make_request_with_result(
+            TOOLS_CONFIG["image_extends"]["api_endpoint"],
+            data,
+            timeout=120
+        )
+
+        return format_response_with_preview(response, "image_extends")
+
+    except Exception as e:
+        return f"❌ Error: {str(e)}", None
+
+
+def image_generator(prompt: str) -> Tuple[str, Optional[str]]:
+    """Generate images using AI from text prompts.
+
+    Args:
+        prompt: Text prompt to generate image from
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    try:
+        if not prompt or not prompt.strip():
+            return "❌ Error: Prompt is required and cannot be empty", None
+
+        client = get_api_client()
+        data = prepare_request_data("image_generator", prompt=prompt.strip())
+
+        response = client.make_request_with_result(
+            TOOLS_CONFIG["image_generator"]["api_endpoint"],
+            data,
+            timeout=120
+        )
+
+        return format_response_with_preview(response, "image_generator")
+
+    except Exception as e:
+        return f"❌ Error: {str(e)}", None
+
+
+# Wrapper functions for Gradio interface
+def remove_bg_wrapper(image_url: str):
+    """Wrapper for remove_bg that returns message and media for Gradio
+
+    Args:
+        image_url: The URL of the image to remove background from. Must be a valid HTTP/HTTPS URL pointing to an image file.
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    message, media_url = remove_bg(image_url)
+    return message, media_url if media_url else None
+
+
+def image_upscaler_wrapper(image_url: str, scale: int):
+    """Wrapper for image_upscaler that returns message and media for Gradio
+
+    Args:
+        image_url: The URL of the image to upscale. Must be a valid HTTP/HTTPS URL pointing to an image file.
+        scale: Scale factor for upscaling. Choose from 2, 4, 8, or 16. Higher values produce larger images but take longer to process.
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    message, media_url = image_upscaler(image_url, scale)
+    return message, media_url if media_url else None
+
+
+def video_upscaler_wrapper(video_url: str):
+    """Wrapper for video_upscaler that returns message and media for Gradio
+
+    Args:
+        video_url: The URL of the video to upscale. Must be a valid HTTP/HTTPS URL pointing to a video file (MP4, AVI, MOV, etc.).
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    message, media_url = video_upscaler(video_url)
+    return message, media_url if media_url else None
+
+
+def image_vectorization_wrapper(image_url: str):
+    """Wrapper for image_vectorization that returns message and media for Gradio
+
+    Args:
+        image_url: The URL of the image to convert to vector format. Must be a valid HTTP/HTTPS URL pointing to an image file.
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    message, media_url = image_vectorization(image_url)
+    return message, media_url if media_url else None
+
+
+def image_extends_wrapper(image_url: str):
+    """Wrapper for image_extends that returns message and media for Gradio
+
+    Args:
+        image_url: The URL of the image to extend. Must be a valid HTTP/HTTPS URL pointing to an image file.
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    message, media_url = image_extends(image_url)
+    return message, media_url if media_url else None
+
+
+def image_generator_wrapper(prompt: str):
+    """Wrapper for image_generator that returns message and media for Gradio
+
+    Args:
+        prompt: Text description of the image to generate. Be descriptive and specific for better results. Example: "A beautiful sunset over mountains with vibrant orange and purple colors".
+
+    Returns:
+        Tuple of (result_message, media_url_for_preview)
+    """
+    message, media_url = image_generator(prompt)
+    return message, media_url if media_url else None
+
+
+# Create Gradio interfaces for each tool
+def create_gradio_app():
+    """Create the main Gradio application with all tools"""
+
+    # Create individual interfaces for each tool
+    remove_bg_interface = gr.Interface(
+        fn=remove_bg_wrapper,
+        inputs=[
+            gr.Textbox(
+                label="Image URL",
+                placeholder="https://example.com/image.jpg",
+                info="Enter the URL of the image to remove background from"
+            )
+        ],
+        outputs=[
+            gr.Textbox(label="Result"),
+            gr.Image(label="Preview", show_label=True)
+        ],
+        title="🎭 Background Removal",
+        description="Remove background from images using AI"
+    )
+
+    image_upscaler_interface = gr.Interface(
+        fn=image_upscaler_wrapper,
+        inputs=[
+            gr.Textbox(
+                label="Image URL",
+                placeholder="https://example.com/image.jpg",
+                info="Enter the URL of the image to upscale"
+            ),
+            gr.Dropdown(
+                choices=[2, 4, 8, 16],
+                value=2,
+                label="Scale Factor",
+                info="Choose the upscaling factor"
+            )
+        ],
+        outputs=[
+            gr.Textbox(label="Result"),
+            gr.Image(label="Preview", show_label=True)
+        ],
+        title="🔍 Image Upscaler",
+        description="Upscale images using AI with specified scale factor"
+    )
+
+    video_upscaler_interface = gr.Interface(
+        fn=video_upscaler_wrapper,
+        inputs=[
+            gr.Textbox(
+                label="Video URL",
+                placeholder="https://example.com/video.mp4",
+                info="Enter the URL of the video to upscale"
+            )
+        ],
+        outputs=[
+            gr.Textbox(label="Result"),
+            gr.Video(label="Preview", show_label=True)
+        ],
+        title="🎬 Video Upscaler",
+        description="Upscale videos using AI"
+    )
+
+    image_vectorization_interface = gr.Interface(
+        fn=image_vectorization_wrapper,
+        inputs=[
+            gr.Textbox(
+                label="Image URL",
+                placeholder="https://example.com/image.jpg",
+                info="Enter the URL of the image to convert to vector format"
+            )
+        ],
+        outputs=[
+            gr.Textbox(label="Result"),
+            gr.Image(label="Preview", show_label=True)
+        ],
+        title="📐 Image Vectorization",
+        description="Convert images to vector format using AI"
+    )
+
+    image_extends_interface = gr.Interface(
+        fn=image_extends_wrapper,
+        inputs=[
+            gr.Textbox(
+                label="Image URL",
+                placeholder="https://example.com/image.jpg",
+                info="Enter the URL of the image to extend"
+            )
+        ],
+        outputs=[
+            gr.Textbox(label="Result"),
+            gr.Image(label="Preview", show_label=True)
+        ],
+        title="🖼️ Image Extension",
+        description="Extend images using AI"
+    )
+
+    image_generator_interface = gr.Interface(
+        fn=image_generator_wrapper,
+        inputs=[
+            gr.Textbox(
+                label="Text Prompt",
+                placeholder="A beautiful sunset over mountains",
+                info="Enter a text description to generate an image",
+                lines=3
+            )
+        ],
+        outputs=[
+            gr.Textbox(label="Result"),
+            gr.Image(label="Preview", show_label=True)
+        ],
+        title="🎨 Image Generator",
+        description="Generate images using AI from text prompts"
+    )
+
+    # Create tabbed interface
+    demo = gr.TabbedInterface(
+        [
+            remove_bg_interface,
+            image_upscaler_interface,
+            video_upscaler_interface,
+            image_vectorization_interface,
+            image_extends_interface,
+            image_generator_interface
+        ],
+        [
+            "Background Removal",
+            "Image Upscaler",
+            "Video Upscaler",
+            "Image Vectorization",
+            "Image Extension",
+            "Image Generator"
+        ],
+        title=GRADIO_CONFIG["title"],
+        theme=GRADIO_CONFIG["theme"]
+    )
+
+    return demo
+
+
+if __name__ == "__main__":
+    # Check for API key
+    if not os.getenv("A1D_API_KEY"):
+        print("❌ Error: A1D_API_KEY environment variable is required")
+        print("Please set your API key: export A1D_API_KEY=your_api_key_here")
+        exit(1)
+
+    # Create and launch the app
+    demo = create_gradio_app()
+
+    # Launch with MCP server enabled
+    demo.launch(
+        mcp_server=True,
+        server_name=GRADIO_CONFIG["server_name"],
+        server_port=GRADIO_CONFIG["server_port"],
+        share=GRADIO_CONFIG["share"]
+    )

config.py

@@ -0,0 +1,84 @@
+"""
+Configuration file for A1D MCP Server
+Contains API endpoints and tool configurations
+"""
+
+import os
+from pathlib import Path
+
+# Load .env file if it exists
+env_file = Path(__file__).parent / '.env'
+if env_file.exists():
+    with open(env_file, 'r') as f:
+        for line in f:
+            line = line.strip()
+            if line and not line.startswith('#') and '=' in line:
+                key, value = line.split('=', 1)
+                os.environ[key.strip()] = value.strip()
+
+# A1D API Configuration
+A1D_API_BASE_URL = "https://api.a1d.ai"
+API_KEY = os.getenv("A1D_API_KEY", "")
+
+# Tool configurations based on the original mcp-server
+TOOLS_CONFIG = {
+    "remove_bg": {
+        "name": "remove_bg",
+        "description": "Remove background from images using AI",
+        "api_endpoint": "/api/remove-bg",
+        "required_params": ["image_url"],
+        "optional_params": [],
+        "param_mapping": {"image_url": "imageUrl"}
+    },
+    "image_upscaler": {
+        "name": "image_upscaler",
+        "description": "Upscale images using AI with specified scale factor",
+        "api_endpoint": "/api/image-upscaler",
+        "required_params": ["image_url"],
+        "optional_params": ["scale"],
+        "default_values": {"scale": 2},
+        "scale_options": [2, 4, 8, 16],
+        "param_mapping": {"image_url": "imageUrl"}
+    },
+    "video_upscaler": {
+        "name": "video_upscaler",
+        "description": "Upscale videos using AI",
+        "api_endpoint": "/api/video-upscaler",
+        "required_params": ["video_url"],
+        "optional_params": [],
+        "param_mapping": {"video_url": "videoUrl"}
+    },
+    "image_vectorization": {
+        "name": "image_vectorization",
+        "description": "Convert images to vector format using AI",
+        "api_endpoint": "/api/image-vectorization",
+        "required_params": ["image_url"],
+        "optional_params": [],
+        "param_mapping": {"image_url": "imageUrl"}
+    },
+    "image_extends": {
+        "name": "image_extends",
+        "description": "Extend images using AI",
+        "api_endpoint": "/api/image-extends",
+        "required_params": ["image_url"],
+        "optional_params": [],
+        "param_mapping": {"image_url": "imageUrl"}
+    },
+    "image_generator": {
+        "name": "image_generator",
+        "description": "Generate images using AI from text prompts",
+        "api_endpoint": "/api/image-generator",
+        "required_params": ["prompt"],
+        "optional_params": []
+    }
+}
+
+# Gradio Configuration
+GRADIO_CONFIG = {
+    "title": "A1D MCP Server - Universal AI Tools",
+    "description": "A powerful MCP server providing AI image and video processing tools",
+    "theme": "default",
+    "share": False,
+    "server_name": "0.0.0.0",
+    "server_port": 7860
+}

mcp_handler.py

@@ -0,0 +1,114 @@
+"""
+MCP Handler for processing requests with API key from headers
+"""
+
+import json
+from typing import Dict, Any, Optional
+from utils import A1DAPIClient, prepare_request_data, format_response_with_preview
+from config import TOOLS_CONFIG
+
+
+def get_api_key_from_headers(headers: Dict[str, str]) -> Optional[str]:
+    """Extract API key from request headers"""
+    # Try different header formats
+    api_key = (headers.get('API_KEY') or 
+               headers.get('api_key') or 
+               headers.get('Api-Key') or
+               headers.get('X-API-Key') or
+               headers.get('x-api-key'))
+    
+    if api_key:
+        print(f"📡 Found API key in headers: {api_key[:8]}...")
+        return api_key
+    
+    print("⚠️ No API key found in headers")
+    return None
+
+
+def process_mcp_request(tool_name: str, params: Dict[str, Any], headers: Dict[str, str]) -> Dict[str, Any]:
+    """Process MCP request with API key from headers"""
+    try:
+        # Get API key from headers
+        api_key = get_api_key_from_headers(headers)
+        if not api_key:
+            return {
+                "error": "API key required. Please provide API_KEY in request headers.",
+                "code": "MISSING_API_KEY"
+            }
+        
+        # Validate tool
+        if tool_name not in TOOLS_CONFIG:
+            return {
+                "error": f"Unknown tool: {tool_name}",
+                "code": "INVALID_TOOL"
+            }
+        
+        print(f"🔧 Processing MCP request for tool: {tool_name}")
+        print(f"📋 Parameters: {params}")
+        
+        # Create API client with header API key
+        client = A1DAPIClient(api_key=api_key)
+        
+        # Prepare request data
+        data = prepare_request_data(tool_name, **params)
+        
+        # Make request with result
+        response = client.make_request_with_result(
+            TOOLS_CONFIG[tool_name]["api_endpoint"],
+            data,
+            timeout=120 if "video" not in tool_name else 300
+        )
+        
+        # Format response
+        message, media_url = format_response_with_preview(response, tool_name)
+        
+        return {
+            "success": True,
+            "message": message,
+            "media_url": media_url,
+            "raw_response": response
+        }
+        
+    except Exception as e:
+        print(f"❌ MCP request error: {str(e)}")
+        return {
+            "error": str(e),
+            "code": "PROCESSING_ERROR"
+        }
+
+
+def create_mcp_tool_functions():
+    """Create MCP tool functions that can handle header-based API keys"""
+    
+    def mcp_remove_bg(image_url: str, headers: Dict[str, str] = None):
+        """Remove background from images using AI (MCP version)"""
+        return process_mcp_request("remove_bg", {"image_url": image_url}, headers or {})
+    
+    def mcp_image_upscaler(image_url: str, scale: int = 2, headers: Dict[str, str] = None):
+        """Upscale images using AI (MCP version)"""
+        return process_mcp_request("image_upscaler", {"image_url": image_url, "scale": scale}, headers or {})
+    
+    def mcp_video_upscaler(video_url: str, headers: Dict[str, str] = None):
+        """Upscale videos using AI (MCP version)"""
+        return process_mcp_request("video_upscaler", {"video_url": video_url}, headers or {})
+    
+    def mcp_image_vectorization(image_url: str, headers: Dict[str, str] = None):
+        """Convert images to vector format using AI (MCP version)"""
+        return process_mcp_request("image_vectorization", {"image_url": image_url}, headers or {})
+    
+    def mcp_image_extends(image_url: str, headers: Dict[str, str] = None):
+        """Extend images using AI (MCP version)"""
+        return process_mcp_request("image_extends", {"image_url": image_url}, headers or {})
+    
+    def mcp_image_generator(prompt: str, headers: Dict[str, str] = None):
+        """Generate images using AI from text prompts (MCP version)"""
+        return process_mcp_request("image_generator", {"prompt": prompt}, headers or {})
+    
+    return {
+        "remove_bg": mcp_remove_bg,
+        "image_upscaler": mcp_image_upscaler,
+        "video_upscaler": mcp_video_upscaler,
+        "image_vectorization": mcp_image_vectorization,
+        "image_extends": mcp_image_extends,
+        "image_generator": mcp_image_generator
+    }

requirements.txt

@@ -0,0 +1,2 @@
+gradio[mcp]>=5.0.0
+requests>=2.31.0

start_server.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""
+Startup script for A1D MCP Server
+"""
+
+import os
+import sys
+from app import create_gradio_app
+
+def main():
+    """Start the A1D MCP Server"""
+    print("🚀 Starting A1D MCP Server...")
+    print("=" * 50)
+    
+    # Check for API key
+    api_key = os.getenv("A1D_API_KEY")
+    if not api_key:
+        print("❌ Error: A1D_API_KEY environment variable is required")
+        print("\n📝 To set your API key:")
+        print("   export A1D_API_KEY=your_api_key_here")
+        print("\n🔗 Get your API key at: https://a1d.ai/home/api")
+        return 1
+    
+    print(f"✅ API key found: {api_key[:8]}...")
+    
+    # Create and launch the app
+    try:
+        demo = create_gradio_app()
+        
+        print("\n🎯 Server Configuration:")
+        print(f"   - Title: {demo.title}")
+        print(f"   - MCP Server: Enabled")
+        print(f"   - Server: http://localhost:7860")
+        print(f"   - MCP Endpoint: http://localhost:7860/gradio_api/mcp/sse")
+        
+        print("\n📋 Available Tools:")
+        from config import TOOLS_CONFIG
+        for tool_name, config in TOOLS_CONFIG.items():
+            print(f"   - {tool_name}: {config['description']}")
+        
+        print("\n🔧 MCP Client Configuration:")
+        print("Add this to your MCP client config:")
+        print("""
+{
+  "mcpServers": {
+    "a1d-gradio": {
+      "command": "npx",
+      "args": [
+        "mcp-remote",
+        "http://localhost:7860/gradio_api/mcp/sse"
+      ]
+    }
+  }
+}
+        """)
+        
+        print("\n🌐 Starting server...")
+        
+        # Launch with MCP server enabled
+        demo.launch(
+            mcp_server=True,
+            server_name="0.0.0.0",
+            server_port=7860,
+            share=False,
+            show_error=True
+        )
+        
+    except KeyboardInterrupt:
+        print("\n👋 Server stopped by user")
+        return 0
+    except Exception as e:
+        print(f"\n❌ Error starting server: {e}")
+        return 1
+
+if __name__ == "__main__":
+    sys.exit(main())

test_app.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+"""
+Test script for A1D MCP Server
+"""
+
+import os
+import sys
+
+def test_imports():
+    """Test if all modules can be imported"""
+    try:
+        import gradio as gr
+        print("✅ Gradio imported successfully")
+        
+        import config
+        print("✅ Config module imported successfully")
+        
+        import utils
+        print("✅ Utils module imported successfully")
+        
+        import app
+        print("✅ App module imported successfully")
+        
+        return True
+    except ImportError as e:
+        print(f"❌ Import error: {e}")
+        return False
+
+def test_config():
+    """Test configuration"""
+    try:
+        from config import TOOLS_CONFIG, GRADIO_CONFIG
+        
+        print(f"✅ Found {len(TOOLS_CONFIG)} tools configured:")
+        for tool_name in TOOLS_CONFIG.keys():
+            print(f"   - {tool_name}")
+        
+        print(f"✅ Gradio config: {GRADIO_CONFIG['title']}")
+        return True
+    except Exception as e:
+        print(f"❌ Config error: {e}")
+        return False
+
+def test_gradio_app():
+    """Test Gradio app creation"""
+    try:
+        # Set a dummy API key for testing
+        os.environ['A1D_API_KEY'] = 'test_key_for_demo'
+        
+        from app import create_gradio_app
+        demo = create_gradio_app()
+        
+        print("✅ Gradio app created successfully")
+        print(f"✅ App title: {demo.title}")
+        
+        return True
+    except Exception as e:
+        print(f"❌ Gradio app error: {e}")
+        return False
+
+def test_tool_functions():
+    """Test individual tool functions"""
+    try:
+        # Set a dummy API key for testing
+        os.environ['A1D_API_KEY'] = 'test_key_for_demo'
+        
+        from app import remove_bg, image_upscaler, image_generator
+        
+        # Test with invalid inputs to check validation
+        result = remove_bg("invalid_url")
+        if "Invalid image URL format" in result:
+            print("✅ URL validation working")
+        
+        result = image_upscaler("invalid_url", 3)
+        if "Invalid image URL format" in result:
+            print("✅ Image upscaler validation working")
+        
+        result = image_generator("")
+        if "Prompt is required" in result:
+            print("✅ Prompt validation working")
+        
+        return True
+    except Exception as e:
+        print(f"❌ Tool function error: {e}")
+        return False
+
+def main():
+    """Run all tests"""
+    print("🧪 Testing A1D MCP Server...")
+    print("=" * 50)
+    
+    tests = [
+        ("Imports", test_imports),
+        ("Configuration", test_config),
+        ("Gradio App", test_gradio_app),
+        ("Tool Functions", test_tool_functions),
+    ]
+    
+    passed = 0
+    total = len(tests)
+    
+    for test_name, test_func in tests:
+        print(f"\n📋 Testing {test_name}...")
+        if test_func():
+            passed += 1
+            print(f"✅ {test_name} test passed")
+        else:
+            print(f"❌ {test_name} test failed")
+    
+    print("\n" + "=" * 50)
+    print(f"🎯 Test Results: {passed}/{total} tests passed")
+    
+    if passed == total:
+        print("🎉 All tests passed! The MCP server is ready to use.")
+        return 0
+    else:
+        print("⚠️  Some tests failed. Please check the errors above.")
+        return 1
+
+if __name__ == "__main__":
+    sys.exit(main())

utils.py

@@ -0,0 +1,308 @@
+"""
+Utility functions for A1D MCP Server
+Handles API calls and data processing
+"""
+
+import requests
+import json
+import os
+import time
+import re
+from typing import Dict, Any, Optional, Tuple
+from config import A1D_API_BASE_URL, API_KEY, TOOLS_CONFIG
+
+
+class A1DAPIClient:
+    """Client for making API calls to A1D services"""
+
+    def __init__(self, api_key: Optional[str] = None):
+        # Try to get API key from multiple sources
+        self.api_key = api_key or self._get_api_key()
+        self.base_url = A1D_API_BASE_URL
+        self.session = requests.Session()
+
+        if not self.api_key:
+            raise ValueError(
+                "API key is required. Set A1D_API_KEY environment variable, pass it directly, or provide via MCP header.")
+
+        # Set default headers
+        self.session.headers.update({
+            "Authorization": f"KEY {self.api_key}",
+            "Content-Type": "application/json",
+            "User-Agent": "A1D-MCP-Server/1.0.0"
+        })
+
+    def _get_api_key(self) -> Optional[str]:
+        """Get API key from various sources"""
+        # 1. Environment variable
+        api_key = API_KEY
+        if api_key:
+            return api_key
+
+        # 2. Try to get from Gradio request headers (if available)
+        try:
+            import gradio as gr
+            request = gr.request()
+            if request and hasattr(request, 'headers'):
+                # Check for API_KEY header from MCP client
+                api_key = request.headers.get(
+                    'API_KEY') or request.headers.get('api_key')
+                if api_key:
+                    print(f"📡 Using API key from MCP client header")
+                    return api_key
+        except:
+            pass
+
+        return None
+
+    def make_request(self, endpoint: str, data: Dict[str, Any]) -> Dict[str, Any]:
+        """Make API request to A1D service"""
+        url = f"{self.base_url}{endpoint}"
+
+        # Add source field to all requests
+        request_data = {**data, "source": "mcp"}
+
+        # Print detailed request information
+        print("\n" + "="*60)
+        print("🚀 A1D API REQUEST DEBUG INFO")
+        print("="*60)
+        print(f"📡 URL: {url}")
+        print(f"🔧 Method: POST")
+
+        print(f"\n📋 Headers:")
+        for key, value in self.session.headers.items():
+            # Mask API key for security
+            if key.lower() in ['api_key', 'authorization']:
+                masked_value = f"{value[:8]}..." if len(value) > 8 else "***"
+                print(f"   {key}: {masked_value}")
+            else:
+                print(f"   {key}: {value}")
+
+        print(f"\n📦 Request Body:")
+        print(f"   {json.dumps(request_data, indent=2)}")
+
+        try:
+            print(f"\n⏳ Sending request...")
+            response = self.session.post(url, json=request_data, timeout=30)
+
+            print(f"\n📊 Response Info:")
+            print(f"   Status Code: {response.status_code}")
+            print(f"   Status Text: {response.reason}")
+
+            print(f"\n📋 Response Headers:")
+            for key, value in response.headers.items():
+                print(f"   {key}: {value}")
+
+            print(f"\n📦 Response Body:")
+            try:
+                response_json = response.json()
+                print(f"   {json.dumps(response_json, indent=2)}")
+            except:
+                print(f"   {response.text[:500]}...")
+
+            print("="*60)
+
+            response.raise_for_status()
+            return response.json()
+
+        except requests.exceptions.RequestException as e:
+            print(f"\n❌ Request failed: {str(e)}")
+            print("="*60)
+            raise Exception(f"API request failed: {str(e)}")
+        except json.JSONDecodeError as e:
+            print(f"\n❌ JSON decode failed: {str(e)}")
+            print("="*60)
+            raise Exception(f"Failed to parse API response: {str(e)}")
+
+    def get_task_result(self, task_id: str, timeout: int = 60) -> Dict[str, Any]:
+        """Get task result using SSE endpoint"""
+        url = f"{self.base_url}/api/task/{task_id}/sse"
+
+        print(f"\n🔄 Getting task result...")
+        print(f"📡 SSE URL: {url}")
+        print(f"⏱️ Timeout: {timeout}s")
+
+        headers = {
+            "Authorization": f"KEY {self.api_key}",
+            "Accept": "text/event-stream"
+        }
+
+        try:
+            response = requests.get(
+                url, headers=headers, stream=True, timeout=timeout)
+            response.raise_for_status()
+
+            print(f"📊 SSE Response Status: {response.status_code}")
+
+            # Parse SSE stream
+            for line in response.iter_lines(decode_unicode=True):
+                if line:
+                    print(f"📥 SSE Line: {line}")
+
+                    # Parse SSE data
+                    if line.startswith("data: "):
+                        data_str = line[6:]  # Remove "data: " prefix
+                        if data_str.strip() == "[DONE]":
+                            print("✅ Task completed!")
+                            break
+
+                        try:
+                            data = json.loads(data_str)
+                            print(
+                                f"📦 Parsed data: {json.dumps(data, indent=2)}")
+
+                            # Check if task is completed
+                            status = data.get("status", "").upper()
+                            if (status in ["COMPLETED", "FINISHED", "SUCCESS"] or
+                                "result" in data or
+                                "imageUrl" in data or
+                                "videoUrl" in data or
+                                    "url" in data):
+                                print("✅ Task result received!")
+                                return data
+                            elif status in ["FAILED", "ERROR"]:
+                                raise Exception(
+                                    f"Task failed: {data.get('error', 'Unknown error')}")
+                            else:
+                                print(
+                                    f"⏳ Task status: {data.get('status', 'processing')}")
+
+                        except json.JSONDecodeError:
+                            print(f"⚠️ Could not parse JSON: {data_str}")
+                            continue
+
+            raise Exception("Task did not complete within timeout")
+
+        except requests.exceptions.RequestException as e:
+            print(f"❌ SSE request failed: {str(e)}")
+            raise Exception(f"Failed to get task result: {str(e)}")
+
+    def make_request_with_result(self, endpoint: str, data: Dict[str, Any], timeout: int = 60) -> Dict[str, Any]:
+        """Make API request and wait for result"""
+        # First, make the initial request to get task ID
+        response = self.make_request(endpoint, data)
+
+        if "taskId" not in response:
+            raise Exception("No taskId in response")
+
+        task_id = response["taskId"]
+        print(f"\n🎯 Task ID: {task_id}")
+
+        # Then get the result
+        return self.get_task_result(task_id, timeout)
+
+
+def validate_url(url: str) -> bool:
+    """Validate if the provided string is a valid URL"""
+    import re
+    url_pattern = re.compile(
+        r'^https?://'  # http:// or https://
+        # domain...
+        r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|'
+        r'localhost|'  # localhost...
+        r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'  # ...or ip
+        r'(?::\d+)?'  # optional port
+        r'(?:/?|[/?]\S+)$', re.IGNORECASE)
+    return url_pattern.match(url) is not None
+
+
+def validate_scale(scale: int) -> bool:
+    """Validate scale parameter for image upscaling"""
+    return scale in TOOLS_CONFIG["image_upscaler"]["scale_options"]
+
+
+def prepare_request_data(tool_name: str, **kwargs) -> Dict[str, Any]:
+    """Prepare request data based on tool configuration"""
+    if tool_name not in TOOLS_CONFIG:
+        raise ValueError(f"Unknown tool: {tool_name}")
+
+    config = TOOLS_CONFIG[tool_name]
+    data = {}
+
+    # Add required parameters
+    for param in config["required_params"]:
+        if param not in kwargs:
+            raise ValueError(f"Missing required parameter: {param}")
+
+        # Apply parameter mapping if exists
+        param_mapping = config.get("param_mapping", {})
+        api_param_name = param_mapping.get(param, param)
+        data[api_param_name] = kwargs[param]
+
+    # Add optional parameters with defaults
+    for param in config.get("optional_params", []):
+        if param in kwargs:
+            # Apply parameter mapping if exists
+            param_mapping = config.get("param_mapping", {})
+            api_param_name = param_mapping.get(param, param)
+            data[api_param_name] = kwargs[param]
+        elif param in config.get("default_values", {}):
+            # Apply parameter mapping if exists
+            param_mapping = config.get("param_mapping", {})
+            api_param_name = param_mapping.get(param, param)
+            data[api_param_name] = config["default_values"][param]
+
+    return data
+
+
+def format_response_with_preview(response: Dict[str, Any], tool_name: str) -> Tuple[str, Optional[str]]:
+    """Format API response for display with media preview
+
+    Returns:
+        Tuple of (message, media_url_for_preview)
+    """
+    if "error" in response:
+        return f"❌ Error: {response['error']}", None
+
+    # Handle different response formats
+    result_url = None
+
+    # Check for A1D API specific fields first
+    result_url = (response.get("imageUrl") or
+                  response.get("videoUrl") or
+                  response.get("url"))
+
+    # Then check nested result fields
+    if not result_url and "result" in response:
+        result = response["result"]
+        if isinstance(result, dict):
+            # Try different possible URL fields
+            result_url = (result.get("imageUrl") or
+                          result.get("videoUrl") or
+                          result.get("url") or
+                          result.get("image_url") or
+                          result.get("video_url") or
+                          result.get("output_url"))
+        elif isinstance(result, str) and result.startswith("http"):
+            result_url = result
+
+    # Also check other common fields
+    if not result_url:
+        result_url = (response.get("image_url") or
+                      response.get("video_url") or
+                      response.get("output_url"))
+
+    if result_url:
+        # Determine media type
+        media_type = "image"
+        if any(ext in result_url.lower() for ext in ['.mp4', '.avi', '.mov', '.webm']):
+            media_type = "video"
+
+        message = f"✅ Success! {media_type.title()} generated: {result_url}"
+        return message, result_url
+
+    return f"✅ Task completed successfully for {tool_name}", None
+
+
+def format_response(response: Dict[str, Any], tool_name: str) -> str:
+    """Format API response for display (backward compatibility)"""
+    message, _ = format_response_with_preview(response, tool_name)
+    return message
+
+
+def get_tool_info(tool_name: str) -> Dict[str, Any]:
+    """Get tool configuration information"""
+    if tool_name not in TOOLS_CONFIG:
+        raise ValueError(f"Unknown tool: {tool_name}")
+
+    return TOOLS_CONFIG[tool_name]