NebulaGraph MCP Server

Instructions

Implementation Reference

• src/nebulagraph_mcp_server/server.py:179-188 (handler)

  The handler function for the 'get_space_schema' MCP tool, decorated with @mcp.tool(). It takes a space name and returns its schema by calling the helper resource function.

  @mcp.tool()
  def get_space_schema(space: str) -> str:
      """Get the schema information of the specified space
      Args:
          space: The space to get the schema for
      Returns:
          The schema information of the specified space
      """
      return get_space_schema_resource(space)

• src/nebulagraph_mcp_server/server.py:71-112 (helper)

  Helper function (also registered as MCP resource) that implements the core logic for retrieving and formatting the schema (tags and edges) of a NebulaGraph space.

  @mcp.resource("schema://space/{space}")
  def get_space_schema_resource(space: str) -> str:
      """Get the schema information of the specified space
      Args:
          space: The space to get the schema for
      Returns:
          The schema information of the specified space
      """
      pool = get_connection_pool()
      session = pool.get_session(
          os.getenv("NEBULA_USER", "root"), os.getenv("NEBULA_PASSWORD", "nebula")
      )
  
      try:
          session.execute(f"USE {space}")
          # Get tags
          tags = session.execute("SHOW TAGS").column_values("Name")
          # Get edges
          edges = session.execute("SHOW EDGES").column_values("Name")
  
          schema = f"Space: {space}\n\nTags:\n"
          for tag in tags:
              tag_result = session.execute(f"DESCRIBE TAG {tag}")
              schema += f"\n{tag}:\n"
              # Iterate through all rows
              for i in range(tag_result.row_size()):
                  field = tag_result.row_values(i)
                  schema += f"  - {field[0]}: {field[1]}\n"
  
          schema += "\nEdges:\n"
          for edge in edges:
              edge_result = session.execute(f"DESCRIBE EDGE {edge}")
              schema += f"\n{edge}:\n"
              # Iterate through all rows
              for i in range(edge_result.row_size()):
                  field = edge_result.row_values(i)
                  schema += f"  - {field[0]}: {field[1]}\n"
  
          return schema
      finally:
          session.release()

• src/nebulagraph_mcp_server/server.py:179-188 (registration)

  Registration of the 'get_space_schema' tool using @mcp.tool() decorator.

  @mcp.tool()
  def get_space_schema(space: str) -> str:
      """Get the schema information of the specified space
      Args:
          space: The space to get the schema for
      Returns:
          The schema information of the specified space
      """
      return get_space_schema_resource(space)