read_query
Execute SELECT queries on the MCP Variance Log's SQLite database to analyze chat monitoring data, including interaction types, probabilities, and summaries for insights.
Instructions
Execute a SELECT query on the SQLite database
Schema Reference:
Table: chat_monitoring
Fields:
- log_id (INTEGER PRIMARY KEY)
- timestamp (DATETIME)
- session_id (TEXT)
- user_id (TEXT)
- interaction_type (TEXT)
- probability_class (TEXT: HIGH, MEDIUM, LOW)
- message_content (TEXT)
- response_content (TEXT)
- context_summary (TEXT)
- reasoning (TEXT)
Example:
SELECT timestamp, probability_class, context_summary
FROM chat_monitoring
WHERE probability_class = 'LOW'
LIMIT 5;
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | SELECT SQL query to execute |
Implementation Reference
- src/mcp_variance_log/server.py:268-275 (handler)The handler function for the 'read_query' tool. Validates that the provided argument is a SELECT query, executes it using the database helper, and returns the results as text content.elif name == "read_query": if not arguments or "query" not in arguments: raise ValueError("Missing query argument") query = arguments["query"].strip() if not query.upper().startswith("SELECT"): raise ValueError("Only SELECT queries are allowed") results = db._execute_query(query) return [types.TextContent(type="text", text=str(results))]
- Input JSON Schema for the 'read_query' tool, defining a required 'query' string parameter.inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SELECT SQL query to execute" } }, "required": ["query"] }
- src/mcp_variance_log/server.py:149-183 (registration)Registration of the 'read_query' tool in the list_tools() handler, including name, detailed description with schema reference and example, and input schema.types.Tool( name="read_query", description="""Execute a SELECT query on the SQLite database Schema Reference: Table: chat_monitoring Fields: - log_id (INTEGER PRIMARY KEY) - timestamp (DATETIME) - session_id (TEXT) - user_id (TEXT) - interaction_type (TEXT) - probability_class (TEXT: HIGH, MEDIUM, LOW) - message_content (TEXT) - response_content (TEXT) - context_summary (TEXT) - reasoning (TEXT) Example: SELECT timestamp, probability_class, context_summary FROM chat_monitoring WHERE probability_class = 'LOW' LIMIT 5; """, inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SELECT SQL query to execute" } }, "required": ["query"] } ),
- Core helper method _execute_query in LogDatabase class that performs the actual SQL execution for read_query. Handles both read and write queries, returns list of dicts for SELECT queries.def _execute_query(self, query: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]: """Execute a SQL query and return results as a list of dictionaries""" logger.debug(f"Executing query: {query}") try: with closing(sqlite3.connect(self.db_path)) as conn: conn.row_factory = sqlite3.Row with closing(conn.cursor()) as cursor: if params: cursor.execute(query, params) else: cursor.execute(query) if query.strip().upper().startswith(('INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP', 'ALTER')): conn.commit() affected = cursor.rowcount logger.debug(f"Write query affected {affected} rows") return [{"affected_rows": affected}] results = [dict(row) for row in cursor.fetchall()] logger.debug(f"Read query returned {len(results)} rows") return results except Exception as e: logger.error(f"Database error executing query: {e}") raise