read_query
Execute SELECT queries on a SQLite database that logs statistical variations in conversation structure and unusual events for analysis.
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)Handler logic for the 'read_query' tool within the @server.call_tool() dispatcher. Validates the presence and SELECT prefix of the query argument, executes it via the database helper, and returns the results as TextContent.
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))] - JSON schema defining the input for the 'read_query' tool: an object with a required 'query' string property.
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 returned by the @server.list_tools() handler, including name, detailed description with schema reference, 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"] } ), - Private helper method of LogDatabase class that executes arbitrary SQL queries (read or write), handles parameters, commits writes, and returns results as list of dictionaries. Used by read_query and other tools.
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