mcp_server_standalone.py

# Real MCP Server for Fire Detection
# Copyright (c) 2024 Wild Fire Tracker
# Licensed under MIT License - see LICENSE file for details

import asyncio
import json
import base64
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from app import FireDetectionMCP
import cv2
import numpy as np

# Initialize fire detection
fire_detector = FireDetectionMCP()

# Create MCP server
server = Server("fire-detection-server")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="analyze_image",
            description="Analyze image for fire and smoke detection",
            inputSchema={
                "type": "object",
                "properties": {
                    "image_base64": {
                        "type": "string",
                        "description": "Base64 encoded image"
                    }
                },
                "required": ["image_base64"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "analyze_image":
        try:
            # Decode base64 image
            image_data = base64.b64decode(arguments["image_base64"])
            nparr = np.frombuffer(image_data, np.uint8)
            frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
            
            # Analyze frame
            status_emoji, status_plain, color = fire_detector.analyze_frame(frame)
            
            return [TextContent(
                type="text",
                text=json.dumps({
                    "status": status_emoji,
                    "color": color,
                    "timestamp": fire_detector.last_detection_time
                })
            )]
        except Exception as e:
            return [TextContent(type="text", text=f"Error: {str(e)}")]
    
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as streams:
        await server.run(*streams)

if __name__ == "__main__":
    asyncio.run(main())