execute_query
Run queries on NebulaGraph database to retrieve graph data and insights from specified spaces.
Instructions
Execute a query Args: query: The query to execute space: The space to use Returns: The results of the query
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| space | Yes |
Implementation Reference
- The 'execute_query' tool handler function. Decorated with @mcp.tool(), it connects to NebulaGraph, switches to the specified space, executes the provided query, formats and returns the results or error message.
@mcp.tool() def execute_query(query: str, space: str) -> str: """Execute a query Args: query: The query to execute space: The space to use Returns: The results of the query """ pool = get_connection_pool() session = pool.get_session( os.getenv("NEBULA_USER", "root"), os.getenv("NEBULA_PASSWORD", "nebula") ) try: session.execute(f"USE {space}") result = session.execute(query) if result.is_succeeded(): # Format the query results if result.row_size() > 0: columns = result.keys() output = "Results:\n" output += " | ".join(columns) + "\n" output += "-" * (len(" | ".join(columns))) + "\n" # Iterate through all rows for i in range(result.row_size()): row = result.row_values(i) output += " | ".join(str(val) for val in row) + "\n" return output return "Query executed successfully (no results)" else: return f"Query failed: {result.error_msg()}" finally: session.release() - src/nebulagraph_mcp_server/server.py:190-190 (registration)The @mcp.tool() decorator registers the 'execute_query' function as an MCP tool.
@mcp.tool() - Function signature and docstring define the input schema (query: str, space: str) and output (str) for the tool, used by MCP for validation.
def execute_query(query: str, space: str) -> str: """Execute a query Args: query: The query to execute space: The space to use Returns: The results of the query """