`Python` with `Starlette`

Install `uv`

First confirm that you have Python version 3.13 installed on your system:

server$ python3 --version
# output:
Python 3.13

At the time of writing, the granian, http.client, psycopg, and maybe other packages are incompatible with Python 3.14.

Installing other versions of Python

Do NOT remove the Python that comes with Ubuntu. Ubuntu relies on it.

To install other versions of Python:

server$ sudo add-apt-repository ppa:deadsnakes/ppa
server$ sudo apt update
server$ sudo apt install python3.13 # for example

To choose the version of Python as default:

server$ sudo update-alternatives --install /usr/bin/python python /usr/bin/python3<TAB> 0 # TAB for autocompletion
server$ sudo update-alternatives --config python3
# then type a selection number corresponding to the Python of choice.

See documentation on deadsnakes personal package archive (PPA)

We’ll be using uv to manage Python package and project. It is reputably faster than other Python package management tools you may have used, such as pip, pip-tools, pipx, poetry, pyenv, twine, virtualenv, and others.

server$ curl -LsSf https://astral.sh/uv/install.sh | sh

Confirm that uv is installed:

server$ uv --version

Create project and install dependencies

Create and change into a directory where you want to keep your harnessd project and initialize the project:

server$ mkdir ~/agentic/harnessd
server$ cd ~/agentic/harnessd
server$ uv init --bare

Install the following packages:

server$ uv add granian http.client 'httpx[http2]' starlette pydantic-core uvloop

update and upgrade

To update uv and upgrade all project packages to the latest compatible version:

  server$ uv self update
  server$ uv sync --upgrade

`main.py`

Create and open a file called main.py:

server$ vi main.py

Edit the file to add the following import lines:

from starlette.applications import Starlette
from starlette.routing import Route

import handlers

We import our module handlers, which we will define later.

We create a routing table to hold the URL routing information needed by Starlette and assign it to a global variable routes. We define the route to serve the HTTP POST request with URL endpoint /llmprompt. We route the endpoint to the llmprompt() function. With each route, we specify which HTTP method is allowed for the URL endpoint, in this case the endpoint accepts only an HTTP POST request:

routes = [
    Route('/llmprompt', handlers.llmprompt, methods=['POST']),
]

The function llmprompt() will be implemented in handlers.py later.

For now, staying in main.py, construct a Starlette web server, providing it the routes array.

# must come after route definitions
app = Starlette(routes=routes)

We’re done with main.py. Save and exit the file.

`harnessd.py`

Python’s web server, including Starlette needs a gateway to “plug” HTTP to Python’s Asynchronous Server Gateway Interface (ASGI)–or WSGI, the synchronous version. The uvicorn gateway from the author of Starlette is a popular choice, but unfortunately at the time of writing it supports only HTTP/1.1. We run Starlette over the Granian gateway for its HTTP/2 support, which is interoperable with Starlette thanks to the ASGI standard.

Create a harnessd.py file:

server$ vi harnessd.py

and create an instance of Granian Server to serve as the gateway between HTTP/2 and the Starlette web server:

from granian.server import Server
from pathlib import Path

Server(
    "main:app",
    address="0.0.0.0",
    port=443,
    interface="asgi",
    loop="uvloop",
    log_access=True,
    ssl_cert=Path("/home/ubuntu/agentic/harnessd.crt"),
    ssl_key=Path("/home/ubuntu/agentic/harnessd.key"),
    workers_kill_timeout=1,
    respawn_failed_workers=True,
    respawn_interval=1.0,    
).serve()

`handlers.py`

We implement URL path API handlers in handlers.py:

server$ vi handlers.py

Start the file with the following imports:

from http import HTTPStatus
import httpx
import logging
from starlette.background import BackgroundTask
from starlette.requests import Request
from starlette.responses import Response, StreamingResponse

import runners

We configure the Python standard logger and add a couple of logging functions to print to console results of handling each HTTP request and, in the case of error, return a tuple of HTTP status code and error message to the client:

logger = logging.getLogger("runner")
logging.basicConfig(level=logging.INFO) # Ensures WARNING logs go to standard output
logging.getLogger("httpx").setLevel(logging.WARNING)

def logOk(clientIP: str, runner: str, model: str):
    logger.info(f"{clientIP} | {runner}:{model}")

def logInfo(clientIP: str, runner: str, model: str, errcode: int, msg: str):
    logger.info(f"{errcode} | {clientIP} | {runner}:{model} {msg} |")

def logClientErr(clientIP: str, runner: str, model: str, errcode: HTTPStatus, errmsg: str) -> Response:
    logger.warning(f"{errcode.value} | {clientIP} | {runner}:{model} |")
    return Response(status_code=errcode, content=errmsg)

We next define the handler llmprompt(), which first pick an LLM runner from a ranked list of available runners. Since different runner provides different models, we patch the "model" field of the incoming prompt to the model available at the runner. We use httpx.AsyncClient() to connect to the LLM runner.

asyncClient = httpx.AsyncClient(
    # 5 s timeout on dead destination, but keep SSE streams open indefinitely
    timeout=httpx.Timeout(connect=5.0, read=None, write=None, pool=None),
    http2=True)

llmRunners = runners.LlmRunners()

async def llmprompt(request: Request):

    clientIP = request.client.host if request.client else "Unknown"

    try:
        body = await request.json()
    except Exception:
        return logClientErr(clientIP, "All", "", HTTPStatus.BAD_REQUEST, "Invalid JSON body.")

    model = body["model"]
    
    for runner in llmRunners:
        # patch the "model" field with the model available for this runner, 
        # if specified, else restore original model that came with request.
        body["model"] = model if len(runner.model) == 0 else runner.model

        # send prompt and return response
        

Then we assemble together a request addressed to the runner’s OpenAI-compatible API end point, along with the preloaded API key for that runner, and send it to the LLM runner. If we fail to connect to the LLM runner, we log the error and try the next runner on our list. If connection is successful but the runner returns any kind of error, including being rate limited, we again log the error and try the next runner on our list. If the runner completes the prompt, we log the success and forward the completion to the user. Replace # send prompt and return response with:

        headers = {"Content-Type": "application/json"}
        if runner.apiKey:
            headers["Authorization"] = f"Bearer {runner.apiKey}"
            
        try:
            # async with asyncClient.send( // NOT for send()
            # https://www.python-httpx.org/async/
            # Cannot use `async with`, instead we will close the response manually in
            # a background task after streaming is done (https://stackoverflow.com/a/73736138)
            response = await asyncClient.send(
                # asyncClient.build_request allows us to stream the response safely 
                asyncClient.build_request(
                    method=request.method,
                    url=f"{runner.baseUrl}/v1/chat/completions",
                    headers=headers,
                    json=body,
                ), stream=True)

            # If rate limited or server error, continue advances the iterator
            if response.status_code != HTTPStatus.OK:
                logInfo(clientIP, runner.baseUrl, runner.model, response.status_code, "Server error. Trying another one.")
                await response.aclose() # cleanup: close the stream directly to prevent asyncClient target connection leaks
                continue
            
            # Success! Break the loop and return the streaming response
            logOk(clientIP, runner.baseUrl, runner.model)
            return StreamingResponse(
                response.aiter_raw(),
                background=BackgroundTask(response.aclose),
            )
            
        except httpx.RequestError as e:
            # catch network/connection errors and failover
            logInfo(clientIP, runner.baseUrl, runner.model, HTTPStatus.PROCESSING, 
                f"Connection error: {e}. Trying another one.")
            continue

    # no runner found

If we’ve exhausted all the runners on our list but couldn’t find any to complete the prompt, we log the error and inform the client. Replace # no runner found with:

    # If the for-loop naturally finishes without returning, we are exhausted.
    return logClientErr(clientIP, "None", "", HTTPStatus.TOO_MANY_REQUESTS,
        "All available LLM providers are rate limited or offline.")

We’re done with handlers.py. Save and exit the file.

`runners.py`

Your harness needs to connect to an LLM to perform prompt completion. This part of the tutorial connects the harness to an LLM runner of your choice. If you have a list of available LLM runners, we do a weighted-random round-robin selection to pick a runner to complete each incoming HTTP-request. You can also provide a locally provisioned LLM runner to fall back to in case no cloud-based runners are available. Once you set this feature up, we will be using it for the rest of the term but will not be spending more time on it.

Create runners.py:

server$ vi runners.py

Start the file with the following imports:

from collections.abc import Iterator
from dataclasses import dataclass
import itertools
import math
import os
import random
import sys

Next we define a struct, LlmRunner, to hold the base URL and API key of an LLM runner. LLM runners each hosts a different set of models. Put the string that identifies the model you want to use to the LLM runner in the model property. We will patch the "model" field of user’s prompt with this string before sending it to this runner. If you want the option to run multiple models from the same runner, create a separate LlmRunner for each model. The weight field records the priority you want to give to this runner. The value of the weight should be in the range [0.0, 1.0], with 1.0 being the highest priority. If a model hosted by Runner A has a weight of 0.8 and two models hosted by Runner B each has a weight of 0.5, you may see Runner B selected more frequently in your trace. The weights don’t have to add up to 1.0. Add the following code to your file:

@dataclass
class LlmRunner:
    baseUrl: str
    model: str
    weight: float
    apiKey: str | None = None

The weighted-random round-robin selection is implemented as the iter() method of an Iterator trait. If the selected runner cannot complete the prompt, we move on to the next one on the list. Failover runners are also selected in a weighted-random round-robin manner. Add the following code to create an iterable list of LlmRunners:

class LlmRunners:
    # The iterator that powers the `for runner in llmRunners` loop:
    # to iterate through the runner array randomized per HTTP-request to load balance
    def __iter__(self) -> Iterator[LlmRunner]:
        '''
        Calculate the Efraimidis and Spirakis A-Res scores based on weights
        "Weighted Random Sampling with a Reservoir", Information Processing Letters, 97:5, March 2006.
        Each runner gets a randomized score that, when sorted, determines
        the fallback order for each HTTP-request.
        Each HTTP-request gets a different ordered list, effecting the
        round-robin selection of runners across HTTP-requests.
        '''
        scored = sorted( # sort ascending by score
            self.runners,
            key=lambda r: -math.log(random.random()) / (r.weight if r.weight > 0.0 else 1e-9)
        )
        
        if self.local:
            # lazily chain the local fallback as the last resort
            return itertools.chain(scored, (self.local,))
        
        # If no local fallback, just return the list iterator
        return iter(scored)

    # __init__()

Finally, __init__() initializes the list of runners. The code here lists LLM runners with free tiers. You have to provide your own API key for each runner in an environment variable. If you have subscription to other LLM runners (such as Claude or ChatGPT), you can add them to the list. If you have provisioned for a local LLM as shown in llmprompt-backend, you can assign it to the local variable as your ultimate fallback server. Replace # __init__() comment with:

    def __init__(self):
        self.local = None
        # If you have provisioned a local llama-server, replace None above with:
        #LlmRunner(baseUrl="http://localhost:11433", apiKey=None, model="", weight=0.0)
        
        apiKeys = {
            "cerebras": os.environ.get("CEREBRAS_API_KEY"),
            "groq": os.environ.get("GROQ_API_KEY"),
            "mistral": os.environ.get("MISTRAL_API_KEY"),
            "gemini": os.environ.get("GEMINI_API_KEY"),
            "nvidia": os.environ.get("NVIDIA_API_KEY"),
            #"openrouter": os.environ.get("OPENROUTER_API_KEY"),
        }

        llmRunners = [
            LlmRunner(
                baseUrl="https://api.cerebras.ai", 
                apiKey=apiKeys["cerebras"],
                model="zai-glm-4.7", 
                weight=0.3,
            ),
            LlmRunner(
                baseUrl="https://api.groq.com/openai", 
                apiKey=apiKeys["groq"],
                model="llama-3.1-8b-instant", 
                weight=0.4,
            ),
            LlmRunner(
                baseUrl="https://api.groq.com/openai", 
                apiKey=apiKeys["groq"], 
                model="meta-llama/llama-4-scout-17b-16e-instruct", 
                weight=0.4,
            ),
            LlmRunner(
                baseUrl="https://api.mistral.ai", 
                apiKey=apiKeys["mistral"], 
                model="ministral-8b-latest", 
                weight=0.3,
            ),
            LlmRunner(
                baseUrl="https://api.mistral.ai", 
                apiKey=apiKeys["mistral"], 
                model="open-mistral-nemo", 
                weight=0.2,
            ),
            LlmRunner(
                baseUrl="https://integrate.api.nvidia.com", 
                apiKey=apiKeys["nvidia"], 
                model="meta/llama-3.1-8b-instruct", 
                weight=0.2,
            ),
            LlmRunner(
                baseUrl="https://integrate.api.nvidia.com", 
                apiKey=apiKeys["nvidia"], 
                model="deepseek-ai/deepseek-v4-flash", 
                weight=0.2,
            ),
            LlmRunner(
                baseUrl="https://integrate.api.nvidia.com", 
                apiKey=apiKeys["nvidia"], 
                model="nvidia/nemotron-nano-12b-v2-vl",
                weight= 0.3,
            ),
        ]
        ''' Don't include Open Router due to its low rate limit and variable model availability
        LlmRunner(
            baseUrl="https://openrouter.ai/api",
            apiKey=apiKeys["nvidia"],
            model="openrouter/free", # whichever best model is available and least loaded
            weight= 0.3,
        ),
        '''       

        # loop to filter out runners missing API Key, 
        # faster than list comprehension due to logging side-effects
        self.runners: list[LlmRunner] = []
        for runner in llmRunners:
            if runner.apiKey:
                print(f"Runner {runner.baseUrl} ACTIVE.", file=sys.stderr)
                self.runners.append(runner)
            else:
                print(f"Runner {runner.baseUrl} INACTIVE. No API KEY.", file=sys.stderr)

        if not (self.runners or self.local):
            print("CRITICAL: No API keys found and no local LLM configured. Terminating process.", file=sys.stderr)
            sys.exit(1)

We’re done with runners.py. Save and exit the file.

Test run

To test run your server, launch it from the command line:

server$ sudo su
# You are now root, note the command-line prompt changed from '$' to '#'.
# You can do a lot of harm with root privileges, so be very careful what you do here.
server# source .venv/bin/activate

When you run source .venv/bin/activate, your prompt should change to indicate that you are now operating within a Python virtual environment, for example:

(harnessd) ubuntu@server:/home/ubuntu/agentic/harnessd#

Storing API keys

When we have only one or two environment variables to set, we can do it on the command-line interface (CLI) right before running python3 -u harnessd.py, e.g.,

(harnessd) ubuntu@server:/home/ubuntu/agentic/harnessd# LLM_RUNNER_API_KEY=YOUR_API_KEY python3 -u harnessd.py

When you have multiple API keys, however, this becomes impractical. A not very secure option is to store your API keys in /root/.bashrc or /root/.zshrc ( make sure you don’t push this file to your git repo!):

server# vi /root/.bashrc

Add these to the end of the file and provide your available API keys:

export CEREBRAS_API_KEY=""
export GROQ_API_KEY=""
export GEMINI_API_KEY=""
export MISTRAL_API_KEY=""
export NVIDIA_API_KEY=""
export OPENROUTER_API_KEY=""

export LD_PRELOAD=/usr/lib/x86_64-linux-gnu/libmimalloc.so
export MIMALLOC_ARENA_RESERVE=320MiB

Note that we have also added an environment variable to point to the mimalloc memory allocator we’re using with harnessd. Mimalloc uses memory more efficiently and supports higher concurrency than either the Python memory manager, Linux malloc, or jemalloc, especially for harness architecture workload. After you add these environment variables to your /root/.bashrc, you’d have to do:

server# source /root/.bashrc

to load them into your current shell. Subsequent sudo su will load them automatically.

Then you can just run your harnessd:

(harnessd) ubuntu@server:/home/ubuntu/agentic/harnessd# python3 -u harnessd.py
# Hit ^C to end the test
(harnessd) ubuntu@server:/home/ubuntu/agentic/harnessd# exit
# So that you're no longer root.
server$

`dotenv`

The dotenv library is often used to read .env file. However, common container, serverless, and managed runtime platforms (such as AWS ECS/Fargate and Lambda, Cloudflare Workers, Heroku, Kubernetes, Vercel, etc.), inject configurations directly into process memory, bypassing physical files. We thus import environment variables directly from the parent (shell) process when you run harnessd from the CLI or access them indirectly through systemd service configuration file.

Now return to the back-end cover page to complete a TODO item and test your implementation following the instructions in the subsequent “Testing llmPrompt APIs” section. You must complete ithe TODO to get full credit for this tutorial.

References

Prepared by Tiberiu Vilcu, Wendan Jiang, Alexander Wu, Benjamin Brengman, Ollie Elmgren, Luke Wassink, Mark Wassink, Nowrin Mohamed, Chenglin Li, Xin Jie ‘Joyce’ Liu, Yibo Pi, and Sugih Jamin

Last updated June 23rd, 2026

`Python` with `Starlette`

Cover Page

Back-end Page

Install `uv`

Create project and install dependencies

`main.py`

`harnessd.py`

`handlers.py`

`runners.py`

Test run

Storing API keys

References

Python with Starlette

Cover Page

Back-end Page

Install uv

Create project and install dependencies

main.py

harnessd.py

handlers.py

runners.py

Test run

Storing API keys

References

`Python` with `Starlette`

Install `uv`

`main.py`

`harnessd.py`

`handlers.py`

`runners.py`