get_kvstore_data

Instructions

Implementation Reference

• src/tools/kvstore/data.py:13-84 (handler)

  The GetKvstoreData class (extends BaseTool) with the execute() method that retrieves documents from a Splunk KV Store collection using optional MongoDB-style query filtering. Handles Splunk availability check, optional app context, query execution, error handling, and returns documents with count.

  class GetKvstoreData(BaseTool):
      """
      Retrieve data from a specific KV Store collection.
      """
  
      METADATA = ToolMetadata(
          name="get_kvstore_data",
          description=(
              "Get documents from a KV Store collection with optional MongoDB-style query filtering. Use this "
              "to fetch lookup/configuration data or narrow results by field values.\n\n"
              "Args:\n"
              "    collection (str): Collection name\n"
              "    app (str, optional): App where the collection resides (defaults to current/app context)\n"
              '    query (object, optional): MongoDB-style filter object (e.g., {"status": "active"})\n\n'
              "Outputs: 'documents' array and 'count'.\n"
              "Security: access and results are constrained by the authenticated user's permissions."
          ),
          category="kvstore",
          tags=["kvstore", "data", "query", "storage"],
          requires_connection=True,
      )
  
      async def execute(
          self, ctx: Context, collection: str, app: str | None = None, query: dict | None = None
      ) -> dict[str, Any]:
          """
          Retrieve data from a KV Store collection.
  
          Args:
              collection: Name of the collection to retrieve data from
              app: Optional app name where the collection resides
              query: Optional MongoDB-style query filter
  
          Returns:
              Dict containing retrieved documents
          """
          log_tool_execution("get_kvstore_data", collection=collection, app=app, query=query)
  
          is_available, service, error_msg = self.check_splunk_available(ctx)
  
          if not is_available:
              return self.format_error_response(error_msg)
  
          self.logger.info(f"Retrieving data from KV Store collection: {collection}")
          await ctx.info(f"Retrieving data from KV Store collection: {collection}")
  
          try:
              # Get the collection from the appropriate app context
              if app:
                  kvstore = service.kvstore[app]
              else:
                  kvstore = service.kvstore
  
              collection_obj = kvstore[collection]
  
              # Retrieve data with optional query filter
              if query:
                  documents = collection_obj.data.query(**query)
              else:
                  documents = collection_obj.data.query()
  
              # Convert to list for response
              doc_list = list(documents)
  
              await ctx.info(f"Retrieved {len(doc_list)} documents from collection {collection}")
              return self.format_success_response({"count": len(doc_list), "documents": doc_list})
  
          except Exception as e:
              self.logger.error(f"Failed to retrieve KV Store data: {str(e)}")
              await ctx.error(f"Failed to retrieve KV Store data: {str(e)}")
              return self.format_error_response(str(e))

• src/tools/kvstore/data.py:18-33 (schema)

  ToolMetadata definition for get_kvstore_data: name, description (with args for collection, app, query), category='kvstore', tags, and requires_connection=True.

  METADATA = ToolMetadata(
      name="get_kvstore_data",
      description=(
          "Get documents from a KV Store collection with optional MongoDB-style query filtering. Use this "
          "to fetch lookup/configuration data or narrow results by field values.\n\n"
          "Args:\n"
          "    collection (str): Collection name\n"
          "    app (str, optional): App where the collection resides (defaults to current/app context)\n"
          '    query (object, optional): MongoDB-style filter object (e.g., {"status": "active"})\n\n'
          "Outputs: 'documents' array and 'count'.\n"
          "Security: access and results are constrained by the authenticated user's permissions."
      ),
      category="kvstore",
      tags=["kvstore", "data", "query", "storage"],
      requires_connection=True,
  )

• src/tools/kvstore/__init__.py:6-8 (registration)

  GetKvstoreData is imported from .data and listed in __all__ for the kvstore tools package.

  from .data import GetKvstoreData
  
  __all__ = ["ListKvstoreCollections", "GetKvstoreData", "CreateKvstoreCollection"]

• src/tools/__init__.py:46-47 (registration)

  GetKvstoreData is re-exported in the top-level src/tools/__init__.py via wildcard import from .kvstore and listed in __all__.

  "GetKvstoreData",
  "CreateKvstoreCollection",

• src/tools/workflows/workflow_requirements.py:574-580 (helper)

  get_kvstore_data listed as a tool reference in workflow requirements with description 'Retrieve data from KV Store collections'.

  "get_kvstore_data": "Retrieve data from KV Store collections",
  "list_kvstore_collections": "List all KV Store collections",
  "create_kvstore_collection": "Create new KV Store collections",
  # Workflow Tools
  "list_workflows": "List available workflows",
  "workflow_runner": "Execute workflows by ID",
  # Utility Tools