How to Ingest a PDF into OpenAI Agents SDK with EnConvert

import json
import os
import re
import time

import requests
from agents import Agent, Runner, function_tool

# The Agents SDK reads OPENAI_API_KEY from the environment.
HEADERS = {"X-API-Key": os.environ["ENCONVERT_API_KEY"]}

# Step 1: Upload the PDF (multipart field name is "files")
with open("/path/to/your/technical-manual.pdf", "rb") as f:
    upload = requests.post(
        "https://api.enconvert.com/v2/ingest/files",
        headers=HEADERS,
        files={"files": ("technical-manual.pdf", f, "application/pdf")},
        timeout=60,
    )
upload.raise_for_status()
job_id = upload.json()["job_id"]

# Step 2: Poll until the job completes
while True:
    poll = requests.get(
        f"https://api.enconvert.com/v2/ingest/{job_id}",
        headers=HEADERS,
        timeout=60,
    )
    poll.raise_for_status()
    job = poll.json()
    if job["status"] == "completed":
        output_url = job["output_url"]
        break
    if job["status"] in ("failed", "canceled"):
        raise RuntimeError(
            f"Ingest job {job_id} {job['status']}: {job.get('error_message')}"
        )
    time.sleep(2)

# Step 3: Download the JSONL output (signed URL, plain GET, no API key)
jsonl = requests.get(output_url, timeout=60)
jsonl.raise_for_status()

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


# Step 4: Expose the chunks to the agent as a function tool
@function_tool
def search_manual(query: str, section: str | None = None) -> str:
    """Search the ingested PDF and return relevant passages with their
    section location. Optionally filter by a heading breadcrumb substring.

    Args:
        query: Keywords describing the information needed.
        section: Optional substring of the heading breadcrumb to filter by.
    """
    terms = set(re.findall(r"\w+", query.lower()))
    scored = []

    for chunk in chunks:
        metadata = chunk["metadata"]
        # headings_path is a list of strings (may be empty), e.g. ["Install", "Torque"]
        breadcrumb = " > ".join(metadata["headings_path"])

        if section and section.lower() not in breadcrumb.lower():
            continue

        words = set(re.findall(r"\w+", f"{breadcrumb} {chunk['content']}".lower()))
        score = len(terms & words)
        if score:
            scored.append((score, breadcrumb, metadata["chunk_index"], chunk["content"]))

    if not scored:
        return "No matching passages found."

    scored.sort(key=lambda item: item[0], reverse=True)
    return "\n\n".join(
        f"[{breadcrumb or 'no heading'} | chunk {chunk_index}]\n{content}"
        for _, breadcrumb, chunk_index, content in scored[:3]
    )


# Step 5: Run the agent
agent = Agent(
    name="Manual Assistant",
    instructions=(
        "Answer questions about the technical manual. Use search_manual to find "
        "passages, cite the section each answer comes from, and say so if the "
        "manual does not cover the question."
    ),
    tools=[search_manual],
)

# Replace with a question about your own document
result = Runner.run_sync(agent, "What are the safety requirements before installation?")
print(result.final_output)