2 posts tagged with "MCP"

"""Doubao Vision MCP Server - OpenAI-compatible image analysis via Volcengine Ark."""

import base64
import os
from pathlib import Path

import httpx
from mcp.server.fastmcp import FastMCP

# Config - inject via environment variables, never hardcode
BASE_URL = os.getenv("DOUBAO_BASE_URL", "https://ark.cn-beijing.volces.com/api/v3")
API_KEY = os.getenv("DOUBAO_API_KEY", "")
MODEL = os.getenv("DOUBAO_MODEL", "doubao-1-5-vision-pro-32k-250115")

mcp = FastMCP("doubao-vision")


def _build_image_content(image_source: str) -> dict:
    """Build image content part from URL or local file path."""
    if image_source.startswith(("http://", "https://")):
        return {"type": "image_url", "image_url": {"url": image_source}}
    # Local file: encode to base64 data URI
    path = Path(image_source)
    if not path.exists():
        raise FileNotFoundError(f"Image not found: {image_source}")
    mime_map = {
        ".jpg": "image/jpeg", ".jpeg": "image/jpeg",
        ".png": "image/png", ".gif": "image/gif",
        ".webp": "image/webp",
    }
    mime = mime_map.get(path.suffix.lower(), "image/png")
    data = base64.b64encode(path.read_bytes()).decode()
    return {
        "type": "image_url",
        "image_url": {"url": f"data:{mime};base64,{data}"},
    }


@mcp.tool()
def analyze_image(image_source: str, prompt: str) -> str:
    """Analyze an image using Doubao Vision Pro model.

    Supports remote URLs and local file paths.

    Args:
        image_source: URL or local file path to the image.
        prompt: What to analyze or describe about the image.
    """
    try:
        image_content = _build_image_content(image_source)
    except FileNotFoundError as e:
        return f"Error: {e}"

    messages = [
        {
            "role": "user",
            "content": [
                {"type": "text", "text": prompt},
                image_content,
            ],
        }
    ]

    resp = httpx.post(
        f"{BASE_URL}/chat/completions",
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Content-Type": "application/json",
        },
        json={
            "model": MODEL,
            "messages": messages,
            "max_tokens": 4096,
        },
        timeout=60,
    )

    if resp.status_code != 200:
        return f"API error {resp.status_code}: {resp.text}"

    data = resp.json()
    return data["choices"][0]["message"]["content"]


if __name__ == "__main__":
    mcp.run(transport="stdio")

@mcp.tool()
def generate_text(prompt: str, model: str = "deepseek-chat") -> str:
    """Generate text using specified model. Supports deepseek-chat, qwen-turbo, etc."""
    providers = {
        "deepseek-chat": {"base": "https://api.deepseek.com/v1", "key": os.getenv("DEEPSEEK_API_KEY")},
        "qwen-turbo": {"base": "https://dashscope.aliyuncs.com/compatible-mode/v1", "key": os.getenv("QWEN_API_KEY")},
    }
    cfg = providers.get(model)
    if not cfg:
        return f"Unknown model: {model}"
    # Call OpenAI-compatible endpoint...

@mcp.tool()
def generate_image(prompt: str, size: str = "1024x1024") -> str:
    """Generate image from text prompt using Stable Diffusion WebUI API."""
    resp = httpx.post(
        "http://localhost:7860/sdapi/v1/txt2img",
        json={"prompt": prompt, "width": 1024, "height": 1024},
        timeout=120,
    )
    data = resp.json()
    img_bytes = base64.b64decode(data["images"][0])
    path = f"/tmp/mcp-gen-{hash(prompt)}.png"
    Path(path).write_bytes(img_bytes)
    return f"Image saved to: {path}"

@mcp.tool()
def text_to_speech(text: str, voice: str = "default") -> str:
    """Convert text to speech audio file."""
    resp = httpx.post(
        "https://openspeech.bytedance.com/api/v1/tts",
        headers={"Authorization": f"Bearer;{os.getenv('TTS_API_KEY')}"},
        json={"text": text, "voice": voice, "format": "mp3"},
        timeout=30,
    )
    path = f"/tmp/mcp-tts-{hash(text)}.mp3"
    Path(path).write_bytes(resp.content)
    return f"Audio saved to: {path}"

@mcp.tool()
def query_database(sql: str) -> str:
    """Execute read-only SQL query on internal database. SELECT only."""
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries are allowed."
    resp = httpx.post(
        "http://internal-api.company.com/query",
        json={"sql": sql},
        headers={"Authorization": f"Bearer {os.getenv('INTERNAL_API_KEY')}"},
        timeout=10,
    )
    return resp.json()["result"]

@mcp.tool()
def smart_query(task: str, content: str) -> str:
    """Route task to optimal model based on complexity."""
    route = {
        "classify": ("qwen-turbo", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
        "analyze_image": ("doubao-1-5-vision-pro-32k-250115", "https://ark.cn-beijing.volces.com/api/v3"),
        "deep_reason": ("deepseek-reasoner", "https://api.deepseek.com/v1"),
    }
    model, base = route.get(task, route["classify"])
    # Unified OpenAI-compatible call...

# providers.py - centralized API config management
PROVIDERS = {
    "doubao": {
        "base_url": "https://ark.cn-beijing.volces.com/api/v3",
        "api_key_env": "DOUBAO_API_KEY",
    },
    "deepseek": {
        "base_url": "https://api.deepseek.com/v1",
        "api_key_env": "DEEPSEEK_API_KEY",
    },
}

# Resources: let AI read your internal docs
@mcp.resource("docs://api-spec")
def get_api_spec() -> str:
    return Path("openapi.yaml").read_text()

# Prompts: preset common analysis templates
@mcp.prompt()
def code_review_prompt(code: str) -> str:
    return f"Review this code for security issues and performance:\n\n{code}"

# Local development
mcp.run(transport="stdio")

# Production deployment (remote AI clients access via HTTP)
mcp.run(transport="sse", host="0.0.0.0", port=8080)