执行SQLite查询
(sql: str)
| 35 | |
| 36 | |
| 37 | def _execute_sqlite(sql: str) -> str: |
| 38 | """执行SQLite查询""" |
| 39 | db_path = DB_CONFIG["path"] |
| 40 | |
| 41 | try: |
| 42 | conn = sqlite3.connect(db_path) |
| 43 | conn.row_factory = sqlite3.Row |
| 44 | cursor = conn.cursor() |
| 45 | |
| 46 | cursor.execute(sql) |
| 47 | rows = cursor.fetchall() |
| 48 | |
| 49 | if rows: |
| 50 | result = [dict(row) for row in rows] |
| 51 | output = json.dumps(result, ensure_ascii=False, indent=2) |
| 52 | else: |
| 53 | output = json.dumps({"message": "查询结果为空"}) |
| 54 | |
| 55 | conn.close() |
| 56 | return output |
| 57 | |
| 58 | except sqlite3.Error as e: |
| 59 | error_output = json.dumps({ |
| 60 | "error": f"SQL执行错误: {str(e)}", |
| 61 | "sql": sql |
| 62 | }, ensure_ascii=False) |
| 63 | return error_output |
| 64 | except Exception as e: |
| 65 | error_output = json.dumps({ |
| 66 | "error": f"未知错误: {str(e)}" |
| 67 | }, ensure_ascii=False) |
| 68 | return error_output |
| 69 | |
| 70 | if __name__ == "__main__": |
| 71 | mcp.run() |