How to Ingest Markdown into OpenAI File Search with EnConvert

import os
import json
import re
import time

import requests
from openai import OpenAI

headers = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}

# Step 1: Upload the Markdown file to EnConvert (multipart field name is "files")
with open("/path/to/your/runbook.md", "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=headers,
        files={"files": ("runbook.md", f, "text/markdown")},
        timeout=120,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 2: Poll until the job completes
deadline = time.monotonic() + 900
while True:
    poll = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=headers,
        timeout=30,
    )
    poll.raise_for_status()
    job = poll.json()
    if job["status"] == "completed":
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(
            f"Ingest job {job_id} {job['status']}: {job.get('error_message')}"
        )
    if time.monotonic() > deadline:
        raise TimeoutError(f"Job {job_id} is still running; resume polling later")
    time.sleep(5)

# Step 3: Download the JSONL output
# The signed URL expires after 15 minutes. Plain GET, no API key.
jsonl = requests.get(job["output_url"], timeout=120)
jsonl.raise_for_status()

# One chunk per line. Split on "\n" only: the text can hold other Unicode line breaks.
chunks = [
    json.loads(line)
    for line in jsonl.content.decode("utf-8").split("\n")
    if line.strip()
]
print(f"Loaded {len(chunks)} chunks")

# Step 4: Create an OpenAI vector store
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
vector_store = client.vector_stores.create(name="runbook-sections")


def slug(text: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")[:60] or "untitled"


# Step 5: Upload each chunk as its own plain text file (a long section gives several chunks)
for chunk in chunks:
    metadata = chunk["metadata"]
    section = metadata.get("section", "")
    chunk_index = metadata["chunk_index"]
    filename = f"{chunk_index:03d}-{slug(section)}.txt"

    uploaded = client.files.create(
        file=(filename, chunk["content"].encode("utf-8")),
        purpose="assistants",
    )

    # Attributes are what the search filter reads. They accept only strings,
    # numbers and booleans, so the headings_path list is joined into one string.
    indexed = client.vector_stores.files.create_and_poll(
        vector_store_id=vector_store.id,
        file_id=uploaded.id,
        attributes={
            "section": section[:500],
            "headings_path": " > ".join(metadata.get("headings_path", []))[:500],
            "chunk_index": chunk_index,
        },
        # Chunks are already section-sized: the largest static size and no
        # overlap keep each file as one piece instead of being re-split
        chunking_strategy={
            "type": "static",
            "static": {
                "max_chunk_size_tokens": 4096,
                "chunk_overlap_tokens": 0,
            },
        },
    )
    if indexed.status != "completed":
        raise RuntimeError(f"OpenAI could not index {filename}: {indexed.last_error}")

print(f"Vector store ready: {len(chunks)} chunk files indexed")

# Step 6: Query with file search, restricted to one section by an attribute filter
response = client.responses.create(
    model="gpt-6-astra",
    input="How do I roll back a failed deployment?",
    tools=[
        {
            "type": "file_search",
            "vector_store_ids": [vector_store.id],
            "filters": {
                "type": "eq",
                "key": "section",
                # Replace with a heading from your own Markdown file
                "value": "Rollback procedure",
            },
        }
    ],
)
print(response.output_text)