MCP Standards

Instructions

Implementation Reference

• src/mcp_standards/server.py:336-372 (handler)

  Main handler function implementing the search_episodes tool. Performs full-text search (FTS) on the episodes table using SQLite, with fallback to LIKE search. Returns matching episodes with metadata.
  async def _search_episodes(self, query: str, limit: int = 10) -> Dict[str, Any]: """Search episodes using FTS""" try: with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row # Use FTS search if available, fallback to LIKE try: cursor = conn.execute(""" SELECT e.id, e.name, e.content, e.source, e.timestamp, rank FROM episodes_search JOIN episodes e ON episodes_search.rowid = e.id WHERE episodes_search MATCH ? ORDER BY rank LIMIT ? """, (query, limit)) except sqlite3.OperationalError: # Fallback to simple search cursor = conn.execute(""" SELECT id, name, content, source, timestamp FROM episodes WHERE name LIKE ? OR content LIKE ? ORDER BY timestamp DESC LIMIT ? """, (f"%{query}%", f"%{query}%", limit)) episodes = [dict(row) for row in cursor.fetchall()] return { "success": True, "query": query, "results": episodes, "count": len(episodes) } except Exception as e: return {"success": False, "error": str(e)}
• src/mcp_standards/server.py:144-155 (schema)

  JSON Schema definition for the search_episodes tool input parameters: query (required string), limit (optional integer default 10).
  Tool( name="search_episodes", description="Search knowledge episodes", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, "limit": {"type": "integer", "description": "Max results", "default": 10}, }, "required": ["query"], }, ),
• src/mcp_standards/server.py:253-258 (registration)

  Tool dispatch/registration in the main call_tool handler: matches tool name and calls the _search_episodes implementation.
  elif name == "search_episodes": results = await self._search_episodes( arguments["query"], arguments.get("limit", 10) ) return [TextContent(type="text", text=json.dumps(results))]
• src/mcp_standards/enhanced_server.py:411-445 (handler)

  Identical handler implementation in the enhanced server variant.
  async def _search_episodes(self, query: str, limit: int = 10) -> Dict[str, Any]: """Search episodes using FTS""" try: with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row try: cursor = conn.execute(""" SELECT e.id, e.name, e.content, e.source, e.timestamp, rank FROM episodes_search JOIN episodes e ON episodes_search.rowid = e.id WHERE episodes_search MATCH ? ORDER BY rank LIMIT ? """, (query, limit)) except sqlite3.OperationalError: cursor = conn.execute(""" SELECT id, name, content, source, timestamp FROM episodes WHERE name LIKE ? OR content LIKE ? ORDER BY timestamp DESC LIMIT ? """, (f"%{query}%", f"%{query}%", limit)) episodes = [dict(row) for row in cursor.fetchall()] return { "success": True, "query": query, "results": episodes, "count": len(episodes) } except Exception as e: return {"success": False, "error": str(e)}
• src/mcp_standards/model_router.py:32-32 (helper)

  Helper classifying search_episodes as SIMPLE task complexity for cost-optimized model routing (e.g., to cheaper models like Gemini Flash).
  "search_episodes": TaskComplexity.SIMPLE,