Quickstart

import json, signal, sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


class Handler(BaseHTTPRequestHandler):
    def answer(self, status, body):
        data = json.dumps(body).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):  # AgentOS polls this until the container is ready
        self.answer(200 if self.path == "/health" else 404, {"status": "ok"})

    def do_POST(self):  # every run of every capability arrives here
        body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}")
        if body.get("capability") != "echo":
            return self.answer(200, {"status": "failed", "error": {
                "code": "unknown_capability", "message": f"no capability {body.get('capability')!r}"}})
        self.answer(200, {"status": "completed", "output": body["input"]})


if __name__ == "__main__":
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))  # stop quickly on `docker stop`
    ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()

FROM python:3.13-slim
WORKDIR /app
COPY app.py .
USER nobody
EXPOSE 8080
CMD ["python", "-u", "app.py"]

name: hello-agent
title: Hello Agent
description: Echoes its input back.
runtime: stateless
port: 8080
capabilities:
  echo:
    description: Returns the input unchanged.
    input:
      type: object
      properties: {message: {type: string}}
      required: [message]
    output: {type: object}
    timeout: 30s