UVF IT: Simple Async Task Inbox Clone

Welcome to another official UVF IT guide! In this quick build you’ll create a lightweight clone of ReplySlip – an async task inbox that lets AI agents post jobs and retrieve results.

We’ll use Python 3, FastAPI, and a local SQLite DB, all free‑tier. In ~45 minutes you’ll have a runnable service you can extend with your own agents.

Before you start

You need Python 3.9+ and pip. The clone skips authentication and schema validation – the real product has those features.
SQLite stores data locally; it won’t survive a container restart without a volume.
FastAPI’s auto‑docs are handy, but remember this is a development server – not production‑ready.

Clone Architecture

Phase 1: Scaffold Project

First we isolate our work in a virtual environment so dependencies don’t clash with other Python projects. Then we install FastAPI and Uvicorn, the lightweight ASGI server that will host our API. This gives us a clean starting point for the inbox service.

With the environment ready, we also create a folder structure that will hold our code and a simple requirements.txt for reproducibility. All subsequent commands will run inside this isolated shell.

Running the server now will show a blank FastAPI app – later phases will fill in the routes and data handling.

python3 -m venv .venv
source .venv/bin/activate
pip install --upgrade pip
pip install fastapi uvicorn
mkdir -p src
cat > requirements.txt <<'EOF'
fastapi
uvicorn[standard]
EOF

Phase 2: Define Data Model

We need a tiny persistence layer, so we’ll wrap SQLite with a helper that creates the table if it doesn’t exist and offers simple CRUD functions. The helper lives in src/db.py and is imported by our models.

Next, a Pydantic model describes the shape of a task: an id, a payload string, and an optional result. FastAPI will use this model to validate incoming JSON and to serialize responses, keeping our API contract explicit and type‑safe.

src/models.py

import sqlite3
from pathlib import Path
from pydantic import BaseModel

DB_PATH = Path(__file__).parent / 'tasks.db'

# Ensure DB and table exist
def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.execute('''
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                payload TEXT NOT NULL,
                result TEXT
            )
        ''')
        conn.commit()

init_db()

class Task(BaseModel):
    id: int | None = None
    payload: str
    result: str | None = None

# Simple helper functions
def insert_task(payload: str) -> int:
    with sqlite3.connect(DB_PATH) as conn:
        cur = conn.execute('INSERT INTO tasks (payload) VALUES (?)', (payload,))
        conn.commit()
        return cur.lastrowid

def get_task(task_id: int) -> dict | None:
    with sqlite3.connect(DB_PATH) as conn:
        cur = conn.execute('SELECT id, payload, result FROM tasks WHERE id = ?', (task_id,))
        row = cur.fetchone()
        return {'id': row[0], 'payload': row[1], 'result': row[2]} if row else None

Phase 3: Core API Endpoints

In this step we expose two tiny HTTP endpoints using FastAPI. The POST /tasks endpoint receives a JSON payload, stores it in a SQLite table, and returns the newly created row's ID. The GET /tasks/{id} endpoint looks up that ID and returns the stored payload, letting you verify that persistence works.

main.py

import sqlite3
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

# Simple SQLite helper – creates table on first import
conn = sqlite3.connect('tasks.db', check_same_thread=False)
cur = conn.cursor()
cur.execute('CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT)')
conn.commit()

class TaskIn(BaseModel):
    payload: str

class TaskOut(BaseModel):
    id: int
    payload: str

@app.post('/tasks', response_model=TaskOut)
def create_task(task: TaskIn):
    cur.execute('INSERT INTO tasks (payload) VALUES (?)', (task.payload,))
    conn.commit()
    task_id = cur.lastrowid
    return {'id': task_id, 'payload': task.payload}

@app.get('/tasks/{task_id}', response_model=TaskOut)
def read_task(task_id: int):
    cur.execute('SELECT id, payload FROM tasks WHERE id = ?', (task_id,))
    row = cur.fetchone()
    if not row:
        raise HTTPException(status_code=404, detail='Task not found')
    return {'id': row[0], 'payload': row[1]}

Phase 4: Run Locally

Now we start the FastAPI development server. Uvicorn will watch for code changes and serve interactive Swagger UI at /docs, where you can manually test the endpoints without leaving your browser.

uvicorn main:app --reload

Phase 5: Verify & Extend

With the server running, use curl (or the Swagger UI) to create a task and then fetch it back. If the payload you receive matches what you sent, the clone is working. From here you can add more fields, async workers, or authentication to grow the inbox.

# Create a new task
curl -s -X POST http://127.0.0.1:8000/tasks \
  -H 'Content-Type: application/json' \
  -d '{"payload":"hello world"}' | jq .

# Assuming the returned id is 1, fetch it back
curl -s http://127.0.0.1:8000/tasks/1 | jq .

Verifying the Stack

Check	Command / Why it matters
Server starts without errors	uvicorn main:app --reload shows 'Uvicorn running on http://127.0.0.1:8000'
POST /tasks returns JSON with an id	curl -s -X POST http://127.0.0.1:8000/tasks -H 'Content-Type: application/json' -d '{"payload":"test"}' \| jq .id
GET /tasks/{id} returns stored payload	curl -s http://127.0.0.1:8000/tasks/1 \| jq .payload

What’s next?

You now have a functional async inbox. Add a background worker that polls pending rows, runs your AI agent, writes the result, and explore authentication or JSON‑Schema validation to approach the full ReplySlip feature set.