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,