feat: add core mcp tools

solo_server/mcp/personal/tavily_search.py

@@ -0,0 +1,45 @@
+from tavily import Client
+from pydantic import BaseModel
+from litserve.mcp import MCP
+import litserve as ls
+
+class SearchRequest(BaseModel):
+    query: str
+
+class SearchAPI(ls.LitAPI):
+    def setup(self, device):
+        # Initialize Tavily client for web search
+        self.client = Client(api_key="YOUR_TAVILY_API_KEY")
+
+    def decode_request(self, request: SearchRequest):
+        return request.query
+
+    def predict(self, x):
+        # Perform an internet search with Tavily
+        response = self.client.search(
+            query=x,
+            top_k=5,
+            web=True  # enable web search mode
+        )
+        # Format results
+        results = []
+        for hit in response.get("hits", []):
+            results.append({
+                "title": hit.get("title", ""),
+                "snippet": hit.get("snippet", ""),
+                "url": hit.get("url", ""),
+                "score": hit.get("score", 0.0)
+            })
+        return results
+
+    def encode_response(self, output):
+        return {"results": output}
+
+if __name__ == "__main__":
+    api = SearchAPI(
+        mcp=MCP(
+            description="Performs internet search using Tavily Search"
+        )
+    )
+    server = ls.LitServer(api)
+    server.run(port=8000)

solo_server/mcp/personal/timer.py

@@ -0,0 +1,79 @@
+import time
+import threading
+import uuid
+from typing import Optional, Literal, Dict
+from pydantic import BaseModel, Field
+from litserve.mcp import MCP
+import litserve as ls
+
+class TimerRequest(BaseModel):
+    action: Literal['start', 'cancel', 'list'] = Field(
+        ..., description="Action to perform: start a new timer, cancel an existing timer, or list active timers"
+    )
+    hours: Optional[int] = Field(0, ge=0, description="Hours for timer duration (used when starting)")
+    minutes: Optional[int] = Field(0, ge=0, description="Minutes for timer duration (used when starting)")
+    seconds: Optional[int] = Field(0, ge=0, description="Seconds for timer duration (used when starting)")
+    timer_id: Optional[str] = Field(None, description="ID of timer to cancel (used when cancelling)")
+
+class TimerAPI(ls.LitAPI):
+    def setup(self, device):
+        # Initialize storage for active timers
+        self.timers: Dict[str, Dict] = {}
+
+    def decode_request(self, request: TimerRequest):
+        return request
+
+    def predict(self, req: TimerRequest):
+        if req.action == 'start':
+            # Compute total duration in seconds
+            total = req.hours * 3600 + req.minutes * 60 + req.seconds
+            if total <= 0:
+                return {"error": "Duration must be greater than zero."}
+            # Generate a unique timer ID
+            timer_id = str(uuid.uuid4())
+            end_time = time.time() + total
+
+            # Callback to clean up after completion
+            def _complete(tid):
+                print(f"Timer {tid} completed.")
+                self.timers.pop(tid, None)
+
+            # Create and start the timer thread
+            t = threading.Timer(total, _complete, args=(timer_id,))
+            t.start()
+            # Store timer metadata
+            self.timers[timer_id] = {"thread": t, "end_time": end_time}
+            return {"timer_id": timer_id, "message": f"Timer started for {total} seconds."}
+
+        elif req.action == 'cancel':
+            tid = req.timer_id
+            if not tid or tid not in self.timers:
+                return {"error": "Invalid or missing timer_id."}
+            # Cancel and remove
+            self.timers[tid]['thread'].cancel()
+            self.timers.pop(tid, None)
+            return {"timer_id": tid, "message": "Timer cancelled."}
+
+        elif req.action == 'list':
+            # List active timers with remaining time
+            now = time.time()
+            active = []
+            for tid, info in self.timers.items():
+                remaining = max(0, int(info['end_time'] - now))
+                active.append({"timer_id": tid, "remaining_seconds": remaining})
+            return {"active_timers": active}
+
+        else:
+            return {"error": "Unsupported action."}
+
+    def encode_response(self, output):
+        return output
+
+if __name__ == "__main__":
+    api = TimerAPI(
+        mcp=MCP(
+            description="Manages timers: start, cancel, and list active timers"
+        )
+    )
+    server = ls.LitServer(api)
+    server.run(port=8000)

solo_server/mcp/personal/translate.py

@@ -0,0 +1,43 @@
+from transformers import pipeline
+from pydantic import BaseModel
+from litserve.mcp import MCP
+import litserve as ls
+
+class TranslateRequest(BaseModel):
+    text: str = Field(..., description="Text to translate")
+    source_lang: str = Field('en', description="Source language code, e.g., 'en'")
+    target_lang: str = Field('fr', description="Target language code, e.g., 'fr'")
+
+class TranslateAPI(ls.LitAPI):
+    def setup(self, device):
+        # Cache translation pipelines per language pair
+        self.device = device
+        self.translators = {}
+
+    def decode_request(self, request: TranslateRequest):
+        return request
+
+    def predict(self, req: TranslateRequest):
+        pair = f"{req.source_lang}-{req.target_lang}"
+        if pair not in self.translators:
+            model_name = f"Helsinki-NLP/opus-mt-{req.source_lang}-{req.target_lang}"
+            self.translators[pair] = pipeline(
+                "translation", model=model_name, device=self.device
+            )
+        translator = self.translators[pair]
+        result = translator(req.text, max_length=512)
+        # Transformers translation pipeline returns a list of dicts
+        translation = result[0].get('translation_text') or result[0].get('generated_text')
+        return {"translation": translation}
+
+    def encode_response(self, output):
+        return output
+
+if __name__ == "__main__":
+    api = TranslateAPI(
+        mcp=MCP(
+            description="Translates text between specified source and target languages"
+        )
+    )
+    server = ls.LitServer(api)
+    server.run(port=8000)