OpenAI a Python: pracujeme so súbormi, shellom, MCP a audiom

Z Python knižníc potrebujeme openai a httpx.

$ mkdir openai_examples
$ cd openai_examples
$ uv init
$ uv add openai httpx

Práca so súbormi

OpenAI Responses API podporuje prácu s dokumentmi (PDF, DOCX, TXT a ďalšie). Dokumenty môžeme odoslať priamo v base64 formáte, alebo ich najprv nahrať do úložiska OpenAI a následne sa na ne odkazovať cez file_id.

Priame odoslanie PDF

Prvý spôsob je priame zakódovanie PDF do base64 a jeho odoslanie v rámci správy.

Base64 je metóda binárno-textového kódovania, ktorá prevádza binárne dáta (napr. PDF súbory, obrázky) na bezpečný ASCII reťazec zložený z 64 znakov: písmen A–Z, a–z, číslic 0–9 a symbolov + a /.

Vďaka tomu je možné binárne súbory prenášať cez kanály, ktoré podporujú iba text – napríklad HTTP hlavičky, JSON payloads alebo zdrojový kód. Dôležité je vedieť, že base64 nie je šifrovanie ani kompresia: dáta zostávajú čitateľné pre kohokoľvek, kto pozná kódovanie, a výsledný reťazec je približne o 33 % väčší ako pôvodný súbor.

from openai import OpenAI
import httpx
import base64

client = OpenAI()

# 1. Fetch the document
doc_url = "https://eknizky.sk/wp-content/uploads/2019/03/rysava-jalovica-martin-kukucin.pdf"
doc_data = httpx.get(doc_url).content

# 2. Encode to base64 and create data URL
base64_doc = base64.b64encode(doc_data).decode('utf-8')
data_url = f"data:application/pdf;base64,{base64_doc}"

# 3. Create the input with the PDF and the Prompt
messages = [
    {
        "type": "message",
        "role": "user",
        "content": [
            {
                "type": "input_file",
                "filename": "document.pdf",
                "file_data": data_url
            },
            {
                "type": "input_text",
                "text": "Summarize this document."
            }
        ]
    }
]

# 4. Generate the response
response = client.responses.create(
    model="gpt-5-mini",
    input=messages
)

print(response.output_text)

Tento prístup je užitočný pre jednorazové spracovanie dokumentu. Skript najprv pomocou knižnice httpx stiahne PDF súbor z danej URL adresy a uloží jeho binárny obsah do premennej. Následne pomocou modulu base64 zakóduje tieto binárne dáta na base64 formát. Keďže funkcia b64encode vracia späť binárne dáta (bytes), volanie .decode('utf-8') ich prevedie na bežný textový reťazec (string), čo je nevyhnutné pre ďalšie spracovanie a vloženie do textovej štruktúry.

Z tohto reťazca vytvorí tzv. data URL v tvare data:application/pdf;base64,..., čo je formát požadovaný OpenAI API na vloženie súboru priamo do správy. Potom poskladá zoznam správ ( messages), kde kombinuje tento súbor ako input_file a textový prompt ako input_text.

Nakoniec odošle túto štruktúru cez klienta OpenAI pomocou metódy responses.create, ktorá spustí generovanie odpovede modelu na základe poskytnutého dokumentu a inštrukcie. Výsledné zhrnutie alebo odpoveď sa následne vytlačí na výstup.

Nahrávanie súborov do úložiska

Tento krok je predpokladom pre metódu odkazovania cez ID. Namiesto odosielania celého obsahu súboru v každej požiadavke ho raz nahrajeme do cloudového úložiska OpenAI pomocou metódy client.files.create. Funkcia vyžaduje, aby bol súbor otvorený v binárnom režime ( "rb"), čo zabezpečí korektné prečítanie akéhokoľvek typu súboru bez poškodenia encodingu.

Kľúčovým parametrom je purpose, ktorý informuje API, na čo bude súbor určený – hodnota "assistants" umožňuje použitie súboru v rámci Assistants alebo Responses API, zatiaľ čo iné hodnoty slúžia napríklad pre fine-tuning. Po odoslaní requestu OpenAI súbor skontroluje, uloží ho a vráti potvrdenie.

import os
from openai import OpenAI

client = OpenAI()

def upload_pdf(file_path):
    if not os.path.exists(file_path):
        print(f"Error: The file '{file_path}' was not found.")
        return

    try:
        print(f"Uploading '{file_path}'...")
        response = client.files.create(
            file=open(file_path, "rb"),
            purpose="assistants"
        )

        print("Upload successful!")
        print(f"File ID: {response.id}")
        print(f"Filename: {response.filename}")
        print(f"Bytes: {response.bytes}")

        return response.id

    except Exception as e:
        print(f"An error occurred: {e}")


file_to_upload = "neprebudeny.pdf"
file_id = upload_pdf(file_to_upload)
print(f"Uploaded file ID: {file_id}")

V odpovedi získame objekt s metadátami, z ktorých najdôležitejší je file_id. Tento unikátny identifikátor môžeme následne ukladať do databázy a používať ho v ďalších požiadavkách, čím obchádzame potrebu opakovaného nahrávania a šetríme prenosové dáta.

Je dôležité mať na pamäti, že súbory v úložisku OpenAI podliehajú politike uchovávania dát – môžu byť automaticky odstránené po uplynutí určitej doby. Môžeme ich tiež zmazať pomocou API alebo manuálne v dashboarde.

Odkazovanie na nahratý súbor

Ak plánujeme dokument používať opakovane, efektívnejšie je nahrať ho do úložiska OpenAI a odkazovať naň cez file_id.

from openai import OpenAI

client = OpenAI()

# Find the file in your OpenAI storage
# Here we search for the file by name if we don't have the ID saved
target_filename = "neprebudeny.pdf"
all_files = client.files.list()

existing_file = next((f for f in all_files if f.filename == target_filename), None)

if not existing_file:
    print(f"File '{target_filename}' not found. Please upload it first.")
else:
    print(f"Found file: {existing_file.filename} (ID: {existing_file.id})")

    # 2. Use the found File ID in your messages list
    messages = [
        {
            "type": "message",
            "role": "user",
            "content": [
                {
                    "type": "input_file",
                    "file_id": existing_file.id  # No base64 needed
                },
                {
                    "type": "input_text",
                    "text": "What is the main conflict in this story?"
                }
            ]
        }
    ]

    # 3. Generate the response
    response = client.responses.create(
        model="gpt-5-mini",
        input=messages
    )

    print(response.output_text)

Tento spôsob je ideálny pre scenáre, kde pracujeme s rovnakými dokumentmi opakovane, napríklad pri budovaní chatbota nad znalostnou bázou. Skript najprv prejde zoznam nahraných súborov v našom OpenAI účte pomocou client.files.list a vyhľadá konkrétny súbor podľa názvu, čím získa jeho unikátny identifikátor file_id.

Ak je súbor nájdený, toto ID sa použije priamo v štruktúre správy, čo eliminuje potrebu opätovného sťahovania, kódovania do base64 a odosielania celého obsahu súboru pri každom volaní API. Výrazne to šetrí prenosové dáta, znižuje latenciu a optimalizuje využitie kontextového okna modelu.

Spúšťanie shellovských príkazov

OpenAI Responses API umožňuje modelom automaticky navrhovať a spúšťať shellovské príkazy prostredníctvom špecializovaného nástroja shell. Model analyzuje požiadavku používateľa, rozhodne, či je potrebné vykonať systémovú operáciu, sformuluje príkaz a vráti ho na vykonanie. Existujú dva režimy spúšťania: container_auto (bezpečný, izolovaný v Docker kontajneri) a local (priamy prístup k lokálnemu systému, vhodný len pre dôveryhodné prostredia).

Spúšťanie v Docker kontajneri

V režime container_auto všetky príkazy bežia v izolovanom prostredí bez prístupu k hostiteľskému systému. Toto minimalizuje riziko neúmyselného poškodenia dát alebo bezpečnostných incidentov.

from openai import OpenAI

client = OpenAI()

query = "What OS is running in the container? Print the OS version."

# Initial request: ask AI to print OS version using container shell
response = client.responses.create(
    model="gpt-5.2",
    instructions="You are a helpful assistant. Use shell tools when needed.",
    input=[{
        "type": "message",
        "role": "user",
        "content": [{"type": "input_text", "text": query}]
    }],
    tools=[{
        "type": "shell",
        "environment": {"type": "container_auto"}
    }],
)

# Simple agent loop: execute shell calls & return results
while True:
    shell_calls = [item for item in response.output if item.type == "shell_call"]
    if not shell_calls:
        break

    tool_outputs = []
    for item in shell_calls:
        call_id = item.call_id
        for command in item.action.commands:
            print(f"Running in container: {command}")

            import subprocess
            result = subprocess.run(
                command, shell=True, capture_output=True, text=True, timeout=30
            )

            # Format outcome per API spec
            outcome = (
                {"type": "timeout"} if result.returncode is None
                else {"type": "exit", "exit_code": result.returncode}
            )

            tool_outputs.append({
                "call_id": call_id,
                "output": [{
                    "stdout": result.stdout,
                    "stderr": result.stderr,
                    "outcome": outcome,
                }]
            })

    # Send results back, continuing the conversation
    response = client.responses.create(
        model="gpt-5.2",
        previous_response_id=response.id, # Important for conversation continuity
        input=[{
            "type": "shell_call_output",
            "call_id": out["call_id"],
            "output": out["output"]
        } for out in tool_outputs]
    )

print("\nFinal Answer:")
for item in response.output:
    if item.type == "message":
        for content in item.content:
            if content.type == "output_text":
                print(content.text)

Kľúčovým konceptom je tu agent loop – cyklus, v ktorom aplikácia kontroluje, či model nevygeneroval žiadosť o vykonanie príkazu ( shell_call). Ak áno, príkaz sa vykoná, výsledky sa naformátujú podľa špecifikácie API a odošlú späť pomocou previous_response_id, čím sa zachová kontext konverzácie. Cyklus pokračuje, kým model nevráti finálnu odpoveď bez ďalších požiadaviek na shell.

$ uv run python container_shell_example.py
Running in container: cat /etc/os-release
Running in container: uname -a

Final Answer:
The container is running **Ubuntu 22.04.5 LTS (Jammy Jellyfish)**.

From `/etc/os-release`:
- `PRETTY_NAME="Ubuntu 22.04.5 LTS"`
- `VERSION_ID="22.04"`
- `VERSION="22.04.5 LTS (Jammy Jellyfish)"`

Model sa rozhodol použiť dva príkazy pre získanie informácií o systéme: cat /etc/os-release a uname -a.

Spúšťanie lokálneho shellu

Režim local udeľuje modelu priamy prístup k lokálnemu systému, čo umožňuje automatizovať komplexné úlohy ako správa súborov, inštalácia balíkov alebo monitorovanie systémových zdrojov. Tento režim vyžaduje zvýšenú opatrnosť a mal by sa používať iba v dôveryhodnom prostredí s vhodnými oprávneniami.

from openai import OpenAI
from dataclasses import dataclass
import subprocess


@dataclass
class CmdResult:
    stdout: str
    stderr: str
    exit_code: int | None
    timed_out: bool


class ShellExecutor:
    def __init__(self, default_timeout = 60):
        self.default_timeout = default_timeout

    def run(self, cmd, timeout = None):
        t = timeout or self.default_timeout
        p = subprocess.Popen(
            cmd,
            shell=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
        )
        try:
            out, err = p.communicate(timeout=t)
            return CmdResult(out, err, p.returncode, False)
        except subprocess.TimeoutExpired:
            p.kill()
            out, err = p.communicate()
            return CmdResult(out, err, p.returncode, True)


client = OpenAI()
executor = ShellExecutor()

# Initial request - use proper message format
response = client.responses.create(
    model="gpt-5.2",
    instructions="You are a Linux system assistant. Use the shell tool to fulfill requests.",
    input=[
        {
            "type": "message",
            "role": "user",
            "content": [
                {
                    "type": "input_text",
                    "text": "find me the largest pdf file in ~/Documents",
                }
            ],
        }
    ],
    tools=[{"type": "shell", "environment": {"type": "local"}}],
)

# --- AGENT LOOP START ---
while True:
    # Check for shell calls in the output
    shell_calls = [item for item in response.output if item.type == "shell_call"]

    if not shell_calls:
        break  # No more shell calls, exit loop

    tool_outputs = []

    # Execute each shell call
    for item in shell_calls:
        call_id = item.call_id
        commands = item.action.commands
        max_output_length = getattr(item.action, "max_output_length", None)

        # Execute all commands
        for command in commands:
            print(f"AI wants to run: {command}")

            # Execute the command
            result = executor.run(command)

            # Format outcome based on timeout status
            if result.timed_out:
                outcome = {"type": "timeout"}
            else:
                outcome = {"type": "exit", "exit_code": result.exit_code}

            # Format output for the API - NOTE: exit_code goes inside outcome
            output_entry = {
                "stdout": result.stdout,
                "stderr": result.stderr,
                "outcome": outcome,
            }

            # Include max_output_length if provided by the model
            if max_output_length is not None:
                output_entry["max_output_length"] = max_output_length

            tool_outputs.append({"call_id": call_id, "output": [output_entry]})
            print(f"Command finished with outcome: {outcome}")

    # Submit results back to the API with previous_response_id to maintain context
    response = client.responses.create(
        model="gpt-5.2",
        previous_response_id=response.id,
        input=[
            {
                "type": "shell_call_output",
                "call_id": output["call_id"],
                "output": output["output"],
            }
            for output in tool_outputs
        ],
    )
# --- AGENT LOOP END ---

# Print the final natural language response
for item in response.output:
    if item.type == "message":
        for content in item.content:
            if content.type == "output_text":
                print(f"\nFinal Answer:\n{content.text}")

V tomto príklade implementujeme triedu ShellExecutor, ktorá zapuzdruje logiku spúšťania príkazov vrátane obsluhy časových limitov, odchytávania štandardného výstupu a chybových správ. Tento prístup zvyšuje čitateľnosť kódu a uľahčuje testovanie. Princíp agentickej slučky zostáva rovnaký ako v predchádzajúcom príklade – cyklická výmena správ medzi aplikáciou a modelom, kým sa úloha úplne nespracuje.

$ uv run python local_shell_example.py
AI wants to run: find ~/Documents -type f -iname '*.pdf' -printf '%s\t%p\n' 2>/dev/null | sort -nr | head -n 1
Command finished with outcome: {'type': 'exit', 'exit_code': 0}

Final Answer:
Largest PDF in `~/Documents`:

- **/home/jano/Documents/precalculus.pdf** — **11,783,266 bytes** (~11.24 MiB)

Integrácia s MCP

Model Context Protocol (MCP) je otvorený štandard pre pripájanie AI modelov k externým dátovým zdrojom a nástrojom. OpenAI podporuje MCP pomocou nástroja mcp. Ukážeme si pripojenie na DeepWiki MCP server, ktorý poskytuje prístup k vedomostnej báze o MCP.

from openai import OpenAI

client = OpenAI()


def ask_deepwiki(query):
    """Query DeepWiki via MCP and return the answer."""

    resp = client.responses.create(
        model="gpt-5-mini",
        tools=[{
            "type": "mcp",
            "server_label": "deepwiki",
            "server_url": "https://mcp.deepwiki.com/mcp",
            "require_approval": "never",
        }],
        input=query,
    )

    return resp.output_text


query = "What transport protocols does MCP (Model Context Protocol) support?"

answer = ask_deepwiki(query)
print(f"{answer}")

V tomto príklade definujeme funkciu ask_deepwiki, ktorá posiela dotaz na DeepWiki MCP server a vracia odpoveď. V parametri tools špecifikujeme, že chceme použiť nástroj typu mcp, pričom poskytujeme potrebné informácie o serveri, ku ktorému sa chceme pripojiť.

$ uv run deep_wiki_mcp.py
Short answer: MCP supports stdio (standard input/output) and HTTP transports.

Details:
- stdio — used for local development and VS Code extensions (default transport).
- HTTP — used for container/remote deployments (enabled when insecure transports are allowed).
- SSE (Server‑Sent Events) was previously supported but has been removed/disabled (deprecated in Azure MCP Server v0.4.0).

You can select the transport via the server start options (e.g., --transport / ServiceStartOptions.Transport).

Práca s audio súbormi

Moderné Responses API zatiaľ neponúka prácu s audiom. Ukážeme si, ako pracovať s audiom pomocou staršieho API.

OpenAI API ponúka tri hlavné endpointy pre prácu s audiom: transcription (prepísanie hovoreného slova na text), translation (preklad hovoreného obsahu do angličtiny) a text-to-speech (syntéza reči z textového vstupu). Každá z týchto funkcií využíva špecializované modely optimalizované pre danú úlohu a podporuje bežné audio formáty ako MP3, WAV, M4A, FLAC alebo OGG.

Prepis reči na text

Model gpt-4o-transcribe dokáže s vysokou presnosťou prepísať audio nahrávku do čitateľného textu. Podporuje viacero jazykov, rozpoznáva hovorcov a dokáže spracovať aj nahrávky s pozaďovým šumom. Výstupom je štruktúrovaný objekt obsahujúci prepísaný text a metadáta o spracovaní.

from openai import OpenAI

client = OpenAI()

file_name = "speech.mp3"

with open(file_name, "rb") as audio_file:
    transcript = client.audio.transcriptions.create(
        model="gpt-4o-transcribe", file=audio_file
    )

print(transcript.text)

Súbor otvárame v binárnom režime "rb", čo je nevyhnutné pre správne načítanie audio dát. Knižnica následne automaticky spravuje chunking a odoslanie dát do API. Výsledný objekt transcript obsahuje atribút text s finálnym prepisom, ktorý môžeme ďalej spracovávať alebo uložiť.

Preklad audia do angličtiny

Model whisper-1 ponúka špeciálnu funkciu prekladu: dokáže priamo previesť hovorený obsah v cudzom jazyku do anglického textu, bez potreby samostatného kroku prepisu a následného prekladu. Toto je výrazne efektívnejšie a často aj presnejšie, pretože model pracuje s pôvodnou intonáciou a kontextom.

from openai import OpenAI

client = OpenAI()

file_name = "russian.wav"

with open(file_name, "rb") as audio_file:
    transcript = client.audio.translations.create(model="whisper-1", file=audio_file)

print(transcript.text)

Tento prístup je ideálny pre situácie, keď potrebujete rýchlo získať anglický prepis zahraničného podcastu, prednášky alebo hovoru. Model automaticky detekuje zdrojový jazyk a produkuje plynulý anglický text, ktorý zachováva význam originálu.

Generovanie reči z textu

Model gpt-4o-mini-tts umožňuje generovať prirodzene znejúcu reč z ľubovoľného textového vstupu. Podporuje viacero hlasových profilov a nastavení rýchlosti, vďaka čomu môžete prispôsobiť výstup konkrétnemu použitiu – či už ide o čítanie článkov, generovanie audiokníh alebo interaktívne hlasové rozhrania.

from pathlib import Path
import openai

speech_file_path = Path(__file__).parent / "speech.mp3"

message = """
Today it is snowing in Bratislava, Slovakia. The temperature is -2 degrees Celsius.
"""

with openai.audio.speech.with_streaming_response.create(
    model="gpt-4o-mini-tts",
    voice="alloy",
    input=message,
) as response:
    response.stream_to_file(speech_file_path)

print(f"Speech audio saved to: {speech_file_path}")

API ponúka trinásť prednastavených hlasov, vrátane alloy, marin, fable, cedar, nova a shimmer. Vďaka streamingovému prístupu sa audio dáta ukladajú priamo do súboru počas generovania, čo šetrí operačnú pamäť a umožňuje spracovanie aj dlhších textov bez rizika pretečenia bufferov.

Všetky príklady z článku a mnohé ďalšie sú dostupné na GitHub repozitári github.com/janbodnar/Python-AI-Skolenie.