server / client

app.py

@@ -1,74 +1,238 @@
-# -*- coding: utf-8 -*-
- 
-import gradio as gr
-import json
+import asyncio
 import os
-import re
-import requests
-
-
-
-def calc_asteroid_factor(data, weight=1.0):
-    """
-    Calculates the asteroid doom probability.
+import json
+from typing import List, Dict, Any, Union
+from contextlib import AsyncExitStack
 
-    Args:
-        data (object): The data structure returned from NASA
-        weight (float): The weight to apply to the hazard factor.
+import gradio as gr
+from gradio.components.chatbot import ChatMessage
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+from anthropic import Anthropic
+from dotenv import load_dotenv
+
+load_dotenv()
+
+loop = asyncio.new_event_loop()
+asyncio.set_event_loop(loop)
+
+class MCPClientWrapper:
+    def __init__(self):
+        self.session = None
+        self.exit_stack = None
+        self.anthropic = Anthropic()
+        self.tools = []
     
-    Returns:
-        Weighted factor for Near Earth Object impact for the next seven days.    
-    """
-
-    objects = [obj for sublist in data["near_earth_objects"].values() for obj in sublist] 
-
-    num_objects = len(objects)
-
-    hazardous_objects = [obj for sublist in data["near_earth_objects"].values() for obj in sublist if obj["is_potentially_hazardous_asteroid"] == True]
-
-    num_hazardous_objects = len(hazardous_objects)
-
-    return (num_hazardous_objects/num_objects * weight) # hazard factor 
-
-
-def fetch_asteroid_data(url, api_key):
-    """
-    Fetches data from the NASA Near Earth Object Web Service.
+    def connect(self, server_path: str) -> str:
+        return loop.run_until_complete(self._connect(server_path))
     
-    Returns:
-        Near Earth Objects for the next seven days.    
-    """
-
-    request_data = requests.get(
-        url,
-        params={"api_key": api_key},  
-    )
-
-    if request_data.status_code != 200:    
-        return None
-    else:    
-        return json.loads(request_data.content)
+    async def _connect(self, server_path: str) -> str:
+        if self.exit_stack:
+            await self.exit_stack.aclose()
+        
+        self.exit_stack = AsyncExitStack()
+        
+        is_python = server_path.endswith('.py')
+        command = "python" if is_python else "node"
+        
+        server_params = StdioServerParameters(
+            command=command,
+            args=[server_path],
+            env={"PYTHONIOENCODING": "utf-8", "PYTHONUNBUFFERED": "1"}
+        )
+        
+        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
+        self.stdio, self.write = stdio_transport
+        
+        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
+        await self.session.initialize()
+        
+        response = await self.session.list_tools()
+        self.tools = [{ 
+            "name": tool.name,
+            "description": tool.description,
+            "input_schema": tool.inputSchema
+        } for tool in response.tools]
+        
+        tool_names = [tool["name"] for tool in self.tools]
+        return f"Connected to MCP server. Available tools: {', '.join(tool_names)}"
     
-
-def calc_doom_probability():
-    """
-    Calculates the overall doom probability.
+    def process_message(self, message: str, history: List[Union[Dict[str, Any], ChatMessage]]) -> tuple:
+        if not self.session:
+            return history + [
+                {"role": "user", "content": message}, 
+                {"role": "assistant", "content": "Please connect to an MCP server first."}
+            ], gr.Textbox(value="")
+        
+        new_messages = loop.run_until_complete(self._process_query(message, history))
+        return history + [{"role": "user", "content": message}] + new_messages, gr.Textbox(value="")
     
-    Returns:
-       The overall doom probability for the next seven days.    
-    """
-
-    nasa_url = "https://api.nasa.gov/neo/rest/v1/feed"
-    nasa_api_key = os.getenv("NASA_API_KEY") 
+    async def _process_query(self, message: str, history: List[Union[Dict[str, Any], ChatMessage]]):
+        claude_messages = []
+        for msg in history:
+            if isinstance(msg, ChatMessage):
+                role, content = msg.role, msg.content
+            else:
+                role, content = msg.get("role"), msg.get("content")
+            
+            if role in ["user", "assistant", "system"]:
+                claude_messages.append({"role": role, "content": content})
+        
+        claude_messages.append({"role": "user", "content": message})
+        
+        response = self.anthropic.messages.create(
+            model="claude-sonnet-4-20250514",
+            max_tokens=1000,
+            messages=claude_messages,
+            tools=self.tools
+        )
+
+        result_messages = []
+        
+        for content in response.content:
+            if content.type == 'text':
+                result_messages.append({
+                    "role": "assistant", 
+                    "content": content.text
+                })
+                
+            elif content.type == 'tool_use':
+                tool_name = content.name
+                tool_args = content.input
+                
+                result_messages.append({
+                    "role": "assistant",
+                    "content": f"I'll use the {tool_name} tool to help answer your question.",
+                    "metadata": {
+                        "title": f"Using tool: {tool_name}",
+                        "log": f"Parameters: {json.dumps(tool_args, ensure_ascii=True)}",
+                        "status": "pending",
+                        "id": f"tool_call_{tool_name}"
+                    }
+                })
+                
+                result_messages.append({
+                    "role": "assistant",
+                    "content": "```json\n" + json.dumps(tool_args, indent=2, ensure_ascii=True) + "\n```",
+                    "metadata": {
+                        "parent_id": f"tool_call_{tool_name}",
+                        "id": f"params_{tool_name}",
+                        "title": "Tool Parameters"
+                    }
+                })
+                
+                result = await self.session.call_tool(tool_name, tool_args)
+                
+                if result_messages and "metadata" in result_messages[-2]:
+                    result_messages[-2]["metadata"]["status"] = "done"
+                
+                result_messages.append({
+                    "role": "assistant",
+                    "content": "Here are the results from the tool:",
+                    "metadata": {
+                        "title": f"Tool Result for {tool_name}",
+                        "status": "done",
+                        "id": f"result_{tool_name}"
+                    }
+                })
+                
+                result_content = result.content
+                if isinstance(result_content, list):
+                    result_content = "\n".join(str(item) for item in result_content)
+                
+                try:
+                    result_json = json.loads(result_content)
+                    if isinstance(result_json, dict) and "type" in result_json:
+                        if result_json["type"] == "image" and "url" in result_json:
+                            result_messages.append({
+                                "role": "assistant",
+                                "content": {"path": result_json["url"], "alt_text": result_json.get("message", "Generated image")},
+                                "metadata": {
+                                    "parent_id": f"result_{tool_name}",
+                                    "id": f"image_{tool_name}",
+                                    "title": "Generated Image"
+                                }
+                            })
+                        else:
+                            result_messages.append({
+                                "role": "assistant",
+                                "content": "```\n" + result_content + "\n```",
+                                "metadata": {
+                                    "parent_id": f"result_{tool_name}",
+                                    "id": f"raw_result_{tool_name}",
+                                    "title": "Raw Output"
+                                }
+                            })
+                except:
+                    result_messages.append({
+                        "role": "assistant",
+                        "content": "```\n" + result_content + "\n```",
+                        "metadata": {
+                            "parent_id": f"result_{tool_name}",
+                            "id": f"raw_result_{tool_name}",
+                            "title": "Raw Output"
+                        }
+                    })
+                
+                claude_messages.append({"role": "user", "content": f"Tool result for {tool_name}: {result_content}"})
+                next_response = self.anthropic.messages.create(
+                    model="claude-3-5-sonnet-20241022",
+                    max_tokens=1000,
+                    messages=claude_messages,
+                )
+                
+                if next_response.content and next_response.content[0].type == 'text':
+                    result_messages.append({
+                        "role": "assistant",
+                        "content": next_response.content[0].text
+                    })
+
+        return result_messages
+
+client = MCPClientWrapper()
+
+def gradio_interface():
+    with gr.Blocks(title="MCP Doomsweek Client") as demo:
+        gr.Markdown("# MCP Doomsweek Assistant")
+        gr.Markdown("Connect to your MCP Doomsweek Server and chat with the assistant")
+        
+        with gr.Row(equal_height=True):
+            with gr.Column(scale=4):
+                server_path = gr.Textbox(
+                    label="Server Script Path",
+                    placeholder="Enter path to server script (e.g., weather.py)",
+                    value="doomsweek_mcp_server.py"
+                )
+            with gr.Column(scale=1):
+                connect_btn = gr.Button("Connect")
+        
+        status = gr.Textbox(label="Connection Status", interactive=False)
+        
+        chatbot = gr.Chatbot(
+            value=[], 
+            height=500,
+            type="messages",
+            show_copy_button=True,
+            avatar_images=("👤", "🤖")
+        )
+        
+        with gr.Row(equal_height=True):
+            msg = gr.Textbox(
+                label="Your Question",
+                placeholder="Ask about weather or alerts (e.g., What's the weather in New York?)",
+                scale=4
+            )
+            clear_btn = gr.Button("Clear Chat", scale=1)
+        
+        connect_btn.click(client.connect, inputs=server_path, outputs=status)
+        msg.submit(client.process_message, [msg, chatbot], [chatbot, msg])
+        clear_btn.click(lambda: [], None, chatbot)
+        
+    return demo
 
-    if not nasa_api_key:
-        return "ERROR: NASA_API_KEY not set."
-    else: 
-        nasa_data = fetch_asteroid_data(nasa_url, nasa_api_key)
-        if not nasa_data:
-            return "ERROR: Unable to fetch data from NASA API."
-        else:
-            return calc_asteroid_factor(nasa_data, weight=1.0)
-        
-demo = gr.Interface(fn=calc_doom_probability, inputs=None, outputs="text")
-demo.launch(mcp_server=True)
+if __name__ == "__main__":
+    if not os.getenv("ANTHROPIC_API_KEY"):
+        print("Warning: ANTHROPIC_API_KEY not found in environment. Please set it in your .env file.")
+    
+    interface = gradio_interface()
+    interface.launch(debug=True)

doomsweek_mcp_server copy.py

@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+ 
+import gradio as gr
+import json
+import os
+import re
+import requests
+
+
+
+def calc_asteroid_factor(data, weight=1.0):
+    """
+    Calculates the asteroid doom probability.
+
+    Args:
+        data (object): The data structure returned from NASA
+        weight (float): The weight to apply to the hazard factor.
+    
+    Returns:
+        Weighted factor for Near Earth Object impact for the next seven days.    
+    """
+
+    objects = [obj for sublist in data["near_earth_objects"].values() for obj in sublist] 
+
+    num_objects = len(objects)
+
+    hazardous_objects = [obj for sublist in data["near_earth_objects"].values() for obj in sublist if obj["is_potentially_hazardous_asteroid"] == True]
+
+    num_hazardous_objects = len(hazardous_objects)
+
+    return (num_hazardous_objects/num_objects * weight) # hazard factor 
+
+
+def fetch_asteroid_data(url, api_key):
+    """
+    Fetches data from the NASA Near Earth Object Web Service.
+    
+    Returns:
+        Near Earth Objects for the next seven days.    
+    """
+
+    request_data = requests.get(
+        url,
+        params={"api_key": api_key},  
+    )
+
+    if request_data.status_code != 200:    
+        return None
+    else:    
+        return json.loads(request_data.content)
+    
+
+def calc_doom_probability():
+    """
+    Calculates the overall doom probability.
+    
+    Returns:
+       The overall doom probability for the next seven days.    
+    """
+
+    nasa_url = "https://api.nasa.gov/neo/rest/v1/feed"
+    nasa_api_key = os.getenv("NASA_API_KEY") 
+
+    if not nasa_api_key:
+        return "ERROR: NASA_API_KEY not set."
+    else: 
+        nasa_data = fetch_asteroid_data(nasa_url, nasa_api_key)
+        if not nasa_data:
+            return "ERROR: Unable to fetch data from NASA API."
+        else:
+            return calc_asteroid_factor(nasa_data, weight=1.0)
+        
+demo = gr.Interface(fn=calc_doom_probability, inputs=None, outputs="text")
+demo.launch(mcp_server=True)

doomsweek_mcp_server.py

@@ -0,0 +1,85 @@
+# -*- coding: utf-8 -*-
+ 
+from mcp.server.fastmcp import FastMCP
+import gradio as gr
+import io
+import json
+import os
+import re
+import sys
+
+import requests
+
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
+sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
+
+mcp = FastMCP("doomsweek_mcp_server")
+
+
+def calc_asteroid_factor(data, weight=1.0):
+    """
+    Calculates the asteroid doom probability.
+
+    Args:
+        data (object): The data structure returned from NASA
+        weight (float): The weight to apply to the hazard factor.
+    
+    Returns:
+        Weighted factor for Near Earth Object impact for the next seven days.    
+    """
+
+    objects = [obj for sublist in data["near_earth_objects"].values() for obj in sublist] 
+
+    num_objects = len(objects)
+
+    hazardous_objects = [obj for sublist in data["near_earth_objects"].values() for obj in sublist if obj["is_potentially_hazardous_asteroid"] == True]
+
+    num_hazardous_objects = len(hazardous_objects)
+
+    return (num_hazardous_objects/num_objects * weight) # hazard factor 
+
+
+def fetch_asteroid_data(url, api_key):
+    """
+    Fetches data from the NASA Near Earth Object Web Service.
+    
+    Returns:
+        Near Earth Objects for the next seven days.    
+    """
+
+    request_data = requests.get(
+        url,
+        params={"api_key": api_key},  
+    )
+
+    if request_data.status_code != 200:    
+        return None
+    else:    
+        return json.loads(request_data.content)
+    
+@mcp.tool()
+def calc_doom_probability():
+    """
+    Calculates the overall doom probability.
+    
+    Returns:
+       The overall doom probability for the next seven days.    
+    """
+
+    nasa_url = "https://api.nasa.gov/neo/rest/v1/feed"
+    nasa_api_key = os.getenv("NASA_API_KEY") 
+
+    if not nasa_api_key:
+        return "ERROR: NASA_API_KEY not set."
+    else: 
+        nasa_data = fetch_asteroid_data(nasa_url, nasa_api_key)
+        if not nasa_data:
+            return "ERROR: Unable to fetch data from NASA API."
+        else:
+            return calc_asteroid_factor(nasa_data, weight=1.0)
+        
+# demo = gr.Interface(fn=calc_doom_probability, inputs=None, outputs="text")
+# demo.launch(mcp_server=True)
+
+if __name__ == "__main__":
+    mcp.run(transport='stdio')

requirements.txt

@@ -0,0 +1,3 @@
+gradio[mcp] 
+anthropic 
+mcp