Create advanced_mcp_tools.py

advanced_mcp_tools.py

@@ -0,0 +1,259 @@
+import os
+import json
+import aiohttp
+import asyncio
+from typing import Dict, List, Optional, Any
+from datetime import datetime
+import numpy as np
+from openai import AsyncOpenAI
+from pydantic import BaseModel
+import logging
+from pathlib import Path
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class AdvancedMCPTools:
+    def __init__(self):
+        self.memory_file = "agent_memory.json"
+        self.knowledge_base = {}
+        self.load_memory()
+        
+        # Initialize OpenAI client if API key is available
+        self.openai_client = None
+        if os.getenv('OPENAI_API_KEY'):
+            self.openai_client = AsyncOpenAI(api_key=os.getenv('OPENAI_API_KEY'))
+    
+    # ========== Memory Management ==========
+    
+    def load_memory(self):
+        """Load agent's memory from file"""
+        try:
+            if os.path.exists(self.memory_file):
+                with open(self.memory_file, 'r', encoding='utf-8') as f:
+                    self.knowledge_base = json.load(f)
+                logger.info(f"Loaded memory with {len(self.knowledge_base)} entries")
+        except Exception as e:
+            logger.error(f"Error loading memory: {e}")
+            self.knowledge_base = {}
+    
+    async def save_memory(self):
+        """Save agent's memory to file"""
+        try:
+            with open(self.memory_file, 'w', encoding='utf-8') as f:
+                json.dump(self.knowledge_base, f, ensure_ascii=False, indent=2)
+            return {"status": "success", "message": "Memory saved successfully"}
+        except Exception as e:
+            logger.error(f"Error saving memory: {e}")
+            return {"status": "error", "message": str(e)}
+    
+    async def remember(self, key: str, value: Any, metadata: Optional[Dict] = None) -> Dict:
+        """Store information in the agent's memory"""
+        try:
+            self.knowledge_base[key] = {
+                "value": value,
+                "timestamp": datetime.now().isoformat(),
+                "metadata": metadata or {}
+            }
+            await self.save_memory()
+            return {"status": "success", "message": "Information stored in memory"}
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+    
+    async def recall(self, key: str) -> Dict:
+        """Retrieve information from the agent's memory"""
+        try:
+            if key in self.knowledge_base:
+                return {
+                    "status": "success",
+                    "key": key,
+                    "data": self.knowledge_base[key]
+                }
+            return {"status": "not_found", "message": "Key not found in memory"}
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+    
+    # ========== Advanced Tools ==========
+    
+    async def analyze_sentiment(self, text: str) -> Dict:
+        """Analyze sentiment of the given text"""
+        if not self.openai_client:
+            return {"error": "OpenAI API key not configured"}
+            
+        try:
+            response = await self.openai_client.chat.completions.create(
+                model="gpt-3.5-turbo",
+                messages=[
+                    {"role": "system", "content": "You are a sentiment analysis tool. Respond with a single word: 'positive', 'negative', or 'neutral'."},
+                    {"role": "user", "content": f"Analyze the sentiment of this text: {text}"}
+                ],
+                max_tokens=10,
+                temperature=0
+            )
+            
+            sentiment = response.choices[0].message.content.strip().lower()
+            return {"sentiment": sentiment, "text": text}
+        except Exception as e:
+            return {"error": str(e)}
+    
+    async def get_news(self, query: str, language: str = "ar", region: str = "eg", max_results: int = 5) -> List[Dict]:
+        """Get news articles based on a query"""
+        try:
+            async with aiohttp.ClientSession() as session:
+                url = f"https://news.google.com/rss/search"
+                params = {
+                    "q": query,
+                    "hl": language,
+                    "gl": region.upper(),
+                    "ceid": f"{region}:{language}"
+                }
+                
+                async with session.get(url, params=params) as response:
+                    if response.status != 200:
+                        return [{"error": f"Failed to fetch news: {response.status}"}]
+                    
+                    from bs4 import BeautifulSoup
+                    import feedparser
+                    
+                    feed = feedparser.parse(await response.text())
+                    articles = []
+                    
+                    for entry in feed.entries[:max_results]:
+                        articles.append({
+                            "title": entry.title,
+                            "link": entry.link,
+                            "published": entry.published,
+                            "source": entry.source.title if hasattr(entry, 'source') else "Unknown"
+                        })
+                    
+                    return articles
+        except Exception as e:
+            return [{"error": str(e)}]
+    
+    async def generate_image(self, prompt: str, size: str = "1024x1024", quality: str = "standard") -> Dict:
+        """Generate an image using DALL-E"""
+        if not self.openai_client:
+            return {"error": "OpenAI API key not configured"}
+            
+        try:
+            response = await self.openai_client.images.generate(
+                model="dall-e-3",
+                prompt=prompt,
+                size=size,
+                quality=quality,
+                n=1,
+            )
+            
+            return {
+                "status": "success",
+                "url": response.data[0].url,
+                "revised_prompt": response.data[0].revised_prompt
+            }
+        except Exception as e:
+            return {"error": str(e)}
+    
+    async def analyze_image(self, image_url: str) -> Dict:
+        """Analyze an image (requires GPT-4 Vision)"""
+        if not self.openai_client:
+            return {"error": "OpenAI API key not configured"}
+            
+        try:
+            response = await self.openai_client.chat.completions.create(
+                model="gpt-4-vision-preview",
+                messages=[
+                    {
+                        "role": "user",
+                        "content": [
+                            {"type": "text", "text": "Describe this image in detail."},
+                            {
+                                "type": "image_url",
+                                "image_url": {"url": image_url},
+                            },
+                        ],
+                    }
+                ],
+                max_tokens=300,
+            )
+            
+            return {
+                "status": "success",
+                "analysis": response.choices[0].message.content
+            }
+        except Exception as e:
+            return {"error": str(e)}
+    
+    # ========== Knowledge Graph ==========
+    
+    async def add_to_knowledge_graph(self, entity: str, relation: str, target: str) -> Dict:
+        """Add a relationship to the knowledge graph"""
+        try:
+            if "knowledge_graph" not in self.knowledge_base:
+                self.knowledge_base["knowledge_graph"] = {}
+            
+            if entity not in self.knowledge_base["knowledge_graph"]:
+                self.knowledge_base["knowledge_graph"][entity] = {}
+            
+            if relation not in self.knowledge_base["knowledge_graph"][entity]:
+                self.knowledge_base["knowledge_graph"][entity][relation] = []
+            
+            self.knowledge_base["knowledge_graph"][entity][relation].append({
+                "target": target,
+                "timestamp": datetime.now().isoformat()
+            })
+            
+            await self.save_memory()
+            return {"status": "success", "message": "Added to knowledge graph"}
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+    
+    async def query_knowledge_graph(self, entity: str, relation: Optional[str] = None) -> Dict:
+        """Query the knowledge graph"""
+        try:
+            if "knowledge_graph" not in self.knowledge_base:
+                return {"status": "not_found", "message": "Knowledge graph is empty"}
+            
+            if entity not in self.knowledge_base["knowledge_graph"]:
+                return {"status": "not_found", "message": f"Entity '{entity}' not found"}
+            
+            if relation:
+                if relation in self.knowledge_base["knowledge_graph"][entity]:
+                    return {
+                        "status": "success",
+                        "entity": entity,
+                        "relation": relation,
+                        "results": self.knowledge_base["knowledge_graph"][entity][relation]
+                    }
+                else:
+                    return {"status": "not_found", "message": f"Relation '{relation}' not found for entity '{entity}'"}
+            else:
+                return {
+                    "status": "success",
+                    "entity": entity,
+                    "relations": self.knowledge_base["knowledge_graph"][entity]
+                }
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+# Example usage
+async def example_usage():
+    tools = AdvancedMCPTools()
+    
+    # Example: Remember something
+    await tools.remember("user_preference", {"theme": "dark", "language": "ar"})
+    
+    # Example: Recall from memory
+    memory = await tools.recall("user_preference")
+    print("Memory recall:", memory)
+    
+    # Example: Analyze sentiment
+    sentiment = await tools.analyze_sentiment("أنا سعيد جداً بهذا الإنجاز ال��ائع!")
+    print("Sentiment analysis:", sentiment)
+    
+    # Example: Get news
+    news = await tools.get_news("الذكاء الاصطناعي", max_results=3)
+    print("Latest AI news:", news)
+
+if __name__ == "__main__":
+    import asyncio
+    asyncio.run(example_usage())