29 lines
976 B
Python
29 lines
976 B
Python
import os
|
|
from neo4j import GraphDatabase
|
|
|
|
NEO4J_URI = os.getenv("NEO4J_URI", "bolt://neo4j:7687")
|
|
NEO4J_USER = os.getenv("NEO4J_USER", "neo4j")
|
|
NEO4J_PASSWORD = os.getenv("NEO4J_PASSWORD", "")
|
|
|
|
_driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
|
|
|
|
class _GraphDB:
|
|
def send_query(self, cypher: str, params=None):
|
|
try:
|
|
records, _, _ = _driver.execute_query(cypher, parameters=params or {}, database_="neo4j")
|
|
# Normalize to list of dicts
|
|
rows = [dict(r) for r in records]
|
|
return {"status": "success", "query_result": rows}
|
|
except Exception as e:
|
|
return {"status": "error", "error_message": f"{type(e).__name__}: {e}"}
|
|
|
|
graphdb = _GraphDB()
|
|
|
|
|
|
# Lightweight helpers for standardized tool returns (Lesson 4)
|
|
def tool_success(key: str, value):
|
|
return {"status": "success", key: value}
|
|
|
|
def tool_error(message: str):
|
|
return {"status": "error", "error_message": message}
|