NebulaGraph MCP Server

Instructions

Implementation Reference

• src/nebulagraph_mcp_server/server.py:227-240 (handler)

  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)

• src/nebulagraph_mcp_server/server.py:114-156 (helper)

  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()