find_path
Discover connections between two nodes in a NebulaGraph database by identifying paths based on specified depth and space parameters.
Instructions
Find paths between two vertices Args: src: The source vertex ID dst: The destination vertex ID space: The space to use depth: The maximum path depth limit: The maximum number of paths to return Returns: The path results
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| src | Yes | ||
| dst | Yes | ||
| space | Yes | ||
| depth | No | ||
| limit | No |
Implementation Reference
- The 'find_path' tool handler registered with @mcp.tool(). It delegates the execution to the get_path_resource helper function.
@mcp.tool() def find_path(src: str, dst: str, space: str, depth: int = 3, limit: int = 10) -> str: """Find paths between two vertices Args: src: The source vertex ID dst: The destination vertex ID space: The space to use depth: The maximum path depth limit: The maximum number of paths to return Returns: The path results """ return get_path_resource(space, src, dst, depth, limit) - Supporting helper function that performs the actual NebulaGraph query (FIND ALL PATH) to find paths between source and destination vertices and formats the output.
@mcp.resource("path://space/{space}/from/{src}/to/{dst}/depth/{depth}/limit/{limit}") def get_path_resource(space: str, src: str, dst: str, depth: int, limit: int) -> str: """Get the path between two vertices Args: space: The space to use src: The source vertex ID dst: The destination vertex ID depth: The maximum path depth limit: The maximum number of paths to return Returns: The path between the source and destination vertices """ pool = get_connection_pool() session = pool.get_session( os.getenv("NEBULA_USER", "root"), os.getenv("NEBULA_PASSWORD", "nebula") ) try: session.execute(f"USE {space}") query = f"""FIND ALL PATH WITH PROP FROM "{src}" TO "{dst}" OVER * BIDIRECT UPTO {depth} STEPS YIELD PATH AS paths | LIMIT {limit}""" result = session.execute(query) if result.is_succeeded(): # Format the path results if result.row_size() > 0: output = f"Find paths from {src} to {dst}: \n\n" # Iterate through all paths for i in range(result.row_size()): path = result.row_values(i)[ 0 ] # The path should be in the first column output += f"Path {i + 1}:\n{path}\n\n" return output return f"No paths found from {src} to {dst}" else: return f"Query failed: {result.error_msg()}" finally: session.release()