DataPilot MCP Server

Instructions

Implementation Reference

• src/main.py:354-369 (handler)

  MCP tool handler for 'explain_query', registered with @mcp.tool() decorator. Takes a SQL query string and delegates to OpenAIClient for explanation.

  @mcp.tool()
  async def explain_query(query: str, ctx: Context) -> str:
      """Explain what a SQL query does in plain English"""
      await ctx.info(f"Explaining query: {query[:100]}...")
      
      try:
          openai = await get_openai_client()
          explanation = await openai.explain_query(query)
          await ctx.info("Generated query explanation")
          return explanation
          
      except Exception as e:
          logger.error(f"Error explaining query: {str(e)}")
          await ctx.error(f"Failed to explain query: {str(e)}")
          raise

• src/openai_client.py:171-211 (helper)

  OpenAIClient method implementing the core logic: calls OpenAI API with tailored system and user prompts to generate plain English explanation of the SQL query.

  async def explain_query(self, query: str) -> str:
      """Explain what a SQL query does in plain English"""
      
      system_prompt = """
      You are a SQL educator. Explain the provided SQL query in plain English.
      
      Your explanation should:
      - Describe what the query does step by step
      - Explain complex operations in simple terms
      - Mention any important performance considerations
      - Be accessible to both technical and non-technical users
      
      Use clear, friendly language.
      """
      
      user_prompt = f"""
      SQL Query to explain:
      {query}
      
      Please explain what this query does.
      """
      
      try:
          response = await self.client.chat.completions.create(
              model=self.model,
              messages=[
                  {"role": "system", "content": system_prompt},
                  {"role": "user", "content": user_prompt}
              ],
              temperature=0.3,
              max_tokens=600
          )
          
          explanation = response.choices[0].message.content.strip()
          logger.info("Generated query explanation")
          return explanation
          
      except Exception as e:
          logger.error(f"Error explaining query: {str(e)}")
          raise Exception(f"Failed to explain query: {str(e)}")