standalone mcp server script

mcp_server_standalone.py

@@ -0,0 +1,70 @@
+# 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())