DataPilot MCP Server

Instructions

Implementation Reference

• src/main.py:73-102 (handler)

  The primary handler function for the 'execute_sql' tool. Registered via @mcp.tool() decorator. Handles SQL execution using SnowflakeClient, provides user feedback via context, and returns QueryResult.
  @mcp.tool() async def execute_sql(request: SQLQueryRequest, ctx: Context) -> QueryResult: """Execute a SQL query on Snowflake and return results""" await ctx.info(f"Executing SQL query: {request.query[:100]}...") try: client = await get_snowflake_client() result = await client.execute_query( request.query, request.limit, request.warehouse ) if result.success: await ctx.info(f"Query executed successfully. {result.row_count} rows returned.") else: await ctx.error(f"Query failed: {result.error}") return result except Exception as e: logger.error(f"Error executing SQL: {str(e)}") await ctx.error(f"Failed to execute query: {str(e)}") return QueryResult( success=False, data=[], columns=[], row_count=0, error=str(e) )
• src/models.py:17-22 (schema)

  Pydantic model defining the input parameters for the execute_sql tool: query (required), limit, warehouse.
  class SQLQueryRequest(BaseModel): """Request model for SQL query execution""" query: str = Field(..., description="SQL query to execute") limit: Optional[int] = Field(None, description="Maximum number of rows to return") warehouse: Optional[str] = Field(None, description="Warehouse to use for this query")
• src/models.py:44-54 (schema)

  Pydantic model defining the output structure returned by the execute_sql tool.
  class QueryResult(BaseModel): """Result of a SQL query execution""" success: bool data: List[Dict[str, Any]] columns: List[str] row_count: int execution_time_ms: Optional[int] = None query_id: Optional[str] = None warehouse_used: Optional[str] = None error: Optional[str] = None
• src/main.py:73-73 (registration)

  FastMCP @mcp.tool() decorator registers the execute_sql function as an MCP tool.
  @mcp.tool()