← Back to Blog

Use AI Agents in Python — OE Runtime HTTP API

OE Runtime has no Python SDK — and it does not need one. Start it in HTTP server mode and call agents from Python with two lines of requests or httpx. Works with FastAPI, Django, Flask, Airflow, and plain scripts.

Step 1 — Start OE Runtime in Server Mode

OE Runtime's --serve flag turns it into a persistent HTTP API. Start it once and call it from Python as many times as you like.

{
  "llm": {
    "provider": "anthropic",
    "apiKey": "sk-ant-...",
    "model": "claude-opus-5"
  },
  "server": {
    "enabled": true,
    "port": 3333,
    "apiKey": "your-secret-key"
  },
  "connectors": []
}

Then start the server — pick your platform:

# npx (no download needed)
npx -y @openenthrium/oe-runtime@latest --serve --config oe-config.json

# Linux binary
./oe-runtime-linux --serve --config oe-config.json &

# Windows
oe-runtime-win.exe --serve --config oe-config.json

You will see:

🚀  OE Runtime Server  v1.7.5
    Run AI agents via HTTP
Listening  http://localhost:3333

Step 2 — Call It from Python

Use requests (or httpx for async). No special library needed.

import requests

OE_URL    = "http://localhost:3333"
OE_APIKEY = "your-secret-key"

def run_agent(agent_file, params=None):
    resp = requests.post(
        f"{OE_URL}/run-file",
        headers={"x-api-key": OE_APIKEY},
        json={"file": agent_file, "params": params or {}},
        timeout=120,
    )
    resp.raise_for_status()
    return resp.json()

# Run a sales report agent with a param
result = run_agent("./agents/sales-report.yaml", {"company": "Acme Corp"})
print(result["output"])

Use in a FastAPI Route

The HTTP call is just an await with httpx. Drop it into any async FastAPI endpoint:

import httpx
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

OE_URL    = "http://localhost:3333"
OE_APIKEY = "your-secret-key"

class ReportRequest(BaseModel):
    company: str

@app.post("/generate-report")
async def generate_report(req: ReportRequest):
    async with httpx.AsyncClient(timeout=120) as client:
        resp = await client.post(
            f"{OE_URL}/run-file",
            headers={"x-api-key": OE_APIKEY},
            json={"file": "./agents/sales-report.yaml", "params": {"company": req.company}},
        )
        resp.raise_for_status()
        return {"report": resp.json()["output"]}

Use in an Airflow DAG

AI agents fit naturally into data pipelines. Deploy OE Runtime as a sidecar service and call it from any Airflow task:

from airflow.decorators import dag, task
from airflow.utils.dates import days_ago
import requests

OE_URL    = "http://oe-runtime-service:3333"  # sidecar or k8s service
OE_APIKEY = "your-secret-key"

@dag(schedule_interval="@weekly", start_date=days_ago(1))
def weekly_sales_report():

    @task()
    def run_sales_agent():
        resp = requests.post(
            f"{OE_URL}/run-file",
            headers={"x-api-key": OE_APIKEY},
            json={"file": "/agents/weekly-sales.yaml", "params": {}},
            timeout=300,
        )
        resp.raise_for_status()
        return resp.json()["output"]

    run_sales_agent()

weekly_sales_report()

Run an Agent from an Inline YAML String

You do not need a file on disk. Pass the YAML as a string in the request body using the /run endpoint — useful for dynamically generated agents:

import requests

agent_yaml = """
name: Quick Summary
instructions: Summarise the following data concisely.
steps:
  - name: Summarise
    content: Write a 3-bullet executive summary of the input provided.
"""

resp = requests.post(
    "http://localhost:3333/run",
    headers={"x-api-key": "your-secret-key"},
    json={
        "yaml": agent_yaml,
        "params": {},
        "input": "Q1 revenue: $4.2M. Q2 revenue: $5.1M. Top product: Enterprise plan (60%).",
    },
    timeout=60,
)
print(resp.json()["output"])

Why This Is Better Than a Python SDK

OE Runtime's connectors — PostgreSQL, MongoDB, Kafka, S3, Slack, and 40+ others — require native Node.js drivers that have no Python equivalent. By calling the HTTP API instead of a Python wrapper, you get:

• All 40+ connectors — no driver compatibility issues
• Agent chains — auto and manual approval flows work out of the box
• Same YAML agents — the same agent.yaml works from CLI, Node.js SDK, and Python HTTP call
• Any Python version — no dependency conflicts, no native extensions
• Any framework — FastAPI, Django, Flask, Airflow, Prefect, raw scripts

Health Check Before Calling

Add a simple health check at startup to confirm OE Runtime is ready before making agent calls:

import requests, time

def wait_for_oe(url="http://localhost:3333", api_key="your-secret-key", retries=10):
    for _ in range(retries):
        try:
            r = requests.get(f"{url}/health", headers={"x-api-key": api_key}, timeout=3)
            if r.status_code == 200:
                print("OE Runtime ready")
                return
        except requests.exceptions.ConnectionError:
            pass
        time.sleep(1)
    raise RuntimeError("OE Runtime did not start in time")

wait_for_oe()
# now safe to call run_agent(...)

---