Direct SDK/API Quickstart

Upload documents, wait for processing, then search them from your application using the GroundX SDKs or REST API. You can pass the search results to a language model as context for its answers.

To have Claude, Codex, or another agent run these operations, see Install GroundX Agent Harness.

This guide covers:

1. How to get your GroundX API Key
2. Setting up GroundX
3. How to create a new bucket
4. How to use an existing bucket
5. How to upload content to a bucket
6. How to check the status of your upload
7. How to search for content
8. How to use your search results to augment an LLM (i.e. RAG)

Step 1: Getting Your API Key

First, create an account.

Log into the GroundX Dashboard and navigate to API Keys.

Store your API key in the GROUNDX_API_KEY environment variable. Keep it out of source code and chat.

Step 2: Setting Up GroundX

Install the Python or TypeScript SDK:

pip install groundx

Initialize the client using your GROUNDX_API_KEY environment variable:

import os
from groundx import GroundX
client = GroundX(
  api_key=os.environ["GROUNDX_API_KEY"],
)

For other languages, or without an SDK, use cURL or any HTTP client.

Step 3: Creating a New Bucket

GroundX identifies text, graphics, and tables, then converts them into text a language model can use. It adds context from elsewhere in the document to each section. These contextualized sections are called semantic objects.

A bucket stores the semantic objects from your documents and lets you search them with natural language queries. Create a bucket and save its unique ID for uploads and searches. Bucket names do not have to be unique:

# Upload documents to GroundX
response = client.buckets.create(
    name="your_bucket_name",
)
bucket_id = response.bucket.bucket_id

Step 4: Using an Existing Bucket

List the buckets in your GroundX account:

# printing a list of existing buckets
#here, the `list` function is not casting, but rather
#is calling the API which lists all buckets.
buckets = client.buckets.list()
#the response object can be cast to a dictionary for
#legibility
print(buckets.dict())
# The value of `buckets.dict()` will resemble the following:
# {
#   "buckets": [
#     {
#       "bucketId": 1,
#       "created": "2023-10-03T08:59:39Z",
#       "fileCount": 1,
#       "fileSize": "3.1GB",
#       "name": "name",
#       "updated": "2023-10-03T08:59:39Z"
#     }
#     ...
#   ]
# }

Use the bucketId of the bucket you want to upload to or search.

Step 5: Upload Content

The Document Upload API processes local files or documents available at a URL and stores their semantic objects in a bucket.

To upload a local file:

# Upload documents to GroundX
from groundx import GroundX, Document
ingest = client.ingest(
  documents=[
    Document(
      bucket_id=bucket_id,
      file_name="my_file1.txt",
      file_path="/local/path/file1.txt",
      file_type="txt",
      search_data=dict(
        key = "value",
      ),
    )
  ]
)

/v1/ingest/documents/local accepts one file per call, up to 8MB, and does not support callbackUrl/callbackData. For larger files, multiple files in one batch, or callback support, use the client.ingest() SDK method shown above (it uploads via a pre-signed URL behind the scenes) rather than calling this endpoint directly.

To upload a second file, send another request the same way:

curl -X POST https://api.groundx.ai/api/v1/ingest/documents/local \
  -H "X-API-Key: <apiKey>" \
  -H "Content-Type: multipart/form-data" \
  -F blob=@my_file2.pdf \
  -F 'metadata={"bucketId":123,"fileName":"my_file2.pdf","fileType":"pdf"}'

To ingest a document from a URL:

# Upload documents to GroundX
from groundx import GroundX, Document
ingest = client.ingest(
  documents=[
    Document(
      bucket_id=bucket_id,
      file_name="my_file1.txt",
      file_path="https://my.source.url.com/file1.txt",
      file_type="txt",
      search_data=dict(
        key = "value",
      ),
    )
  ]
)

See the Ingest API for all arguments.

Both local and remote ingest return a response with this shape:

{
  "ingest": {
    "processId": "<unique_system_generated_id>",
    "status": "<enumerated_status>"
  }
}

The processId can be used to check the status of the upload.

Step 6: Check the Status of Your Upload

Check processing status:

ingest = client.documents.get_processing_status_by_id(
  process_id=ingest.ingest.process_id
)

Use the processId returned by ingest.

The response has this shape:

{
  "ingest": {
    "processId": "<unique_system_generated_id>",
    "progress": {
      "complete": {
        "documents": [
          {
            "documentId": "<unique_system_generated_id>",
            "fileName": "<given_file_name>",
            "fileSize": "<files_size_total>",
            "fileType": "<file_type>",
            "bucketId": <your_bucket_id>,
            "processId": "<unique_system_generated_id>",
            "sourceUrl": "<document_url>",
            "status": "<enumerated_status>"
          }
        ],
        "total": 1
      }
    },
    "status": "<enumerated_status>"
  }
}

Poll while status is queued, training, or processing. Search after complete; investigate error or cancelled before continuing.

Step 7: Search Your Content

Search the processed documents:

search_response = client.search.content(
  id=bucket_id,
  query=query,
)

Use a bucketId to search one bucket, a groupId for a group of buckets, or a documentId for one document. Set query to your search question.

The response has this shape:

{
  "search": {
    "count": <int_number_of_results>,
    "query": "<your_query>",
    "score": <float_highest_relevance_score_in_results>,
    "text": "<combined_text_of_search_results>",
    "nextToken": "<token_for_next_set_of_results>",
    "results":[
      {
        "documentId": "<unique_system_generated_id>",
        "score": <float_relevance_score_of_result>,
        "searchData": {
            <document_metadata>
        },
        "sourceUrl": "<source_document_url>",
        "suggestedText": "<rewritten_text_for_LLM_completions>",
        "text":  "<original_text_of_result>"
      }
    ]
  }
}

Find groupId and bucketId values in the GroundX Dashboard or through the list APIs.

Step 8: Using Search Results to Augment an LLM

Use search.text as context for your language model. If you build your own context from individual results, use search.results[n].suggestedText.

Read search.text from the response:

llm_text = search_response.search.text

Combine your instructions and GroundX context, then submit them to OpenAI. For a system prompt example and guidance, see Prompting and Integration.

import os
from openai import OpenAI
openai_client = OpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
)
completion = openai_client.chat.completions.create(
    model=openaiModel,
    messages=[
        {
            "role": "system",
            "content": f"""{instruction}
===
{llm_text}
==="""
        },
        {"role": "user", "content": query},
    ],
)
completion = completion.choices[0].message.content

Set openaiModel to your model, instruction to your instructions, llmText to the GroundX context, and query to your question. This is retrieval augmented generation (RAG): the model answers using retrieved document content.

Errors and Rate Limits

HTTP status codes

All GroundX API endpoints use standard HTTP status codes. The two you’re most likely to encounter while integrating are:

For 401, check the X-API-Key header and the key’s resource access. For 400, check the request body and path/query parameters. The response usually describes the validation problem.

Ingest limits

Check file size and page count before uploading. Split documents above the recommended limits to keep searches focused. Batches of 20 leave room for retries and reduce timeout risk on slower connections.

Search limits

Queue or debounce search requests in your application to stay within these targets. If you fan out a single user query to multiple buckets or groups, count each search call separately toward the concurrency limit.

Handling HTTP 429

If you receive a 429 Too Many Requests response, back off and retry with exponential delay: start at 1 second, double on each retry, cap at 60 seconds. Don’t retry immediately.

import time
def search_with_backoff(client, bucket_id, query, max_retries=5):
    delay = 1
    for attempt in range(max_retries):
        try:
            return client.search.content(id=bucket_id, query=query)
        except Exception as e:
            if getattr(e, "status_code", None) == 429 and attempt < max_retries - 1:
                time.sleep(delay)
                delay = min(delay * 2, 60)
            else:
                raise

Next Steps

See Prompting and Integration for guidance on producing consistent answers.