HTTP servers

declarative web endpoints. in 6 lines.

the hello world

FX.StartHTTPServer(
    routes={
        '/':       'Hello, Zef!',
        '/health': ET.JSON(content={'status': 'ok'}),
    },
    port=8000,
) | run

That's a running HTTP server. Two routes. curl localhost:8000 gives you "Hello, Zef!". /health returns JSON.

the server is a VALUE

The argument to FX.StartHTTPServer is just a dict. Routes map paths to values. Each value knows how to become a response.

┌────────────────────────────────────────────┐ │ FX.StartHTTPServer(...) | run │ ├────────────────────────────────────────────┤ │ │ │ routes = { '/': 'hi', │ │ '/health': ET.JSON(...), │ │ '/api': some_handler, │ │ } │ │ │ │ every route value is either: │ │ - a literal (string, ET.JSON, ET.HTML) │ │ - a handler (fn or ZefOp) │ │ │ └────────────────────────────────────────────┘

response types

routes = {
    '/text':     'plain string',
    '/html':     ET.HTML(content='<h1>Hi</h1>'),
    '/json':     ET.JSON(content={'k': 'v'}),
    '/empty':    nil,                                 # 204
    '/redirect': ET.Redirect(url='https://x.com'),    # 302
    '/image':    ET.FromFile(path='/static/a.jpg'), # file
    '/status':   ET.Response(status=418, body='teapot'),
}

HTTP methods

By default, routes answer GET. Specify a method by using a tuple key:

routes = {
    '/users':                list_users_handler,
    ('POST',   '/users'):     create_user_handler,
    ('DELETE', '/users'):     delete_user_handler,
}

HEAD is handled automatically (RFC 9110 §9.3.2) — defined-for-GET routes respond to HEAD with the same status and headers, empty body.

handler functions

A handler is a @zef_function that takes the request and returns a response value:

@zef_function
def list_books(req):
    books = [
        {'title': 'Dune', 'author': 'Frank Herbert'},
        {'title': 'Zef',  'author': 'us'},
    ]
    return ET.JSON(content={'books': books})

FX.StartHTTPServer(
    routes={
        '/':          ET.HTML(content='<h1>Bookshop</h1>'),
        '/api/books': list_books,
    },
    port=8000,
) | run

the request object

Handlers receive a request value. Useful fields:

field	description
`F.path`	URL path (e.g. `/api/users`)
`F.method`	HTTP method (`'GET'`, `'POST'`, …)
`F.headers`	request headers dict
`F.body`	request body (String or Bytes)
`F.client_ip`	client's IP
`F.query`	parsed query string
`F.form_data`	multipart form data (if any)
`F.req_time`	time received

JSON POST body example

@zef_function
def ingest(req):
    data = str(req.body) | parse_json | collect
    print(f'received {len(data)} items')
    return ET.JSON(content={'ok': True, 'n': len(data)})

FX.StartHTTPServer(
    routes={
        '/health':  ET.JSON(content={'status': 'ok'}),
        ('POST', '/ingest'): ingest,
    },
    port=8181,
) | run

import time
while True: time.sleep(1)   # keep process alive

Test:

curl -s -X POST http://127.0.0.1:8181/ingest \
  -H 'Content-Type: application/json' \
  -d '[{"t":1,"v":10},{"t":2,"v":20}]'

server parameters

FX.StartHTTPServer(
    routes={...},                   # OR domains= OR request_handler=
    port=8000,                      # required
    allow_external_requests=False,  # False = localhost only
    allow_restart=False,            # True = replace existing server on port
    certificates=[...],             # for HTTPS, see ch 23
) | run

path variables are not yet supported

You can't do ('GET', '/users/{id}'): handler in routes= — this errors at startup. Workarounds:

Put the parameter in the body
Use query strings and parse F.query
Use request_handler=F.path | match([...]) for manual dispatch

three routing modes

1. `routes=` — domain-agnostic

Responds to any Host header. Good for local dev, single-app containers.

FX.StartHTTPServer(routes={'/': 'hi'}, port=8000) | run

2. `domains=` — multi-domain hosting

FX.StartHTTPServer(
    domains={
        'app.example.com': ET.Domain(routes={'/': ET.HTML(content='App')}),
        'api.example.com': ET.Domain(routes={'/status': ET.JSON(content={'ok': True})}),
    },
    port=443,
    certificates=[...]
) | run

Supports subdomain variables:

domains={
    '{tenant}.myapp.com': ET.Domain(
        routes={'/': F.tenant | apply(lambda t: ET.HTML(content=f'hi {t}'))},
    ),
}

3. `request_handler=` — full control

FX.StartHTTPServer(
    request_handler=F.path | match(
        (Z == '/')                       >> 'Hello',
        (Z | starts_with('/api'))          >> api_handler,
        (Z | starts_with('/users/'))       >> user_handler,
        _                                 >> ET.Response(status=404, body='not found'),
    ),
    port=8000,
) | run

a bigger example — a tiny API

db = FX.CreateDB(type=DictDatabase, persistence='in_memory') | run
db['users'] = []

@zef_function
def list_users(req):
    return ET.JSON(content={'users': list(db['users'])})

@zef_function
def create_user(req):
    data = str(req.body) | parse_json | collect
    users = list(db['users'])
    users.append(data)
    db['users'] = users
    return ET.Response(status=201, body=to_json(data))

@zef_function
def echo(req):
    return ET.JSON(content={
        'method':    req.method,
        'path':      req.path,
        'client_ip': req.client_ip,
    })

FX.StartHTTPServer(
    routes={
        '/':                'Hello',
        '/echo':            echo,
        '/api/users':       list_users,
        ('POST', '/api/users'): create_user,
    },
    port=8000,
) | run

SSE — server-sent events

news = ET.Topic(generate_uid())

FX.StartHTTPServer(
    routes={
        '/': ET.HTML(content="""
            <h1>live news</h1>
            <div id="x"></div>
            <script>
              const src = new EventSource('/events');
              src.onmessage = e => x.innerHTML += '<p>'+e.data+'</p>';
            </script>"""),
        '/events': ET.ServerSentEvents(topic=news),
    },
    port=8000,
) | run

# elsewhere, pushing to all connected clients:
FX.Publish(target=news, content='breaking!') | run

build a tiny todo API

Endpoints:

GET /todos → list all
POST /todos → add a new one from body {"text":"..."}
DELETE /todos → clear all

solution sketch

db = FX.CreateDB(type=DictDatabase, persistence='in_memory') | run
db['items'] = []

@zef_function
def ls(req):   return ET.JSON(content={'items': list(db['items'])})

@zef_function
def add(req):
    it = str(req.body) | parse_json | collect
    items = list(db['items'])
    items.append(it)
    db['items'] = items
    return ET.Response(status=201, body='ok')

@zef_function
def clear(req):
    db['items'] = []
    return ET.Response(status=204)

FX.StartHTTPServer(routes={
    '/todos': ls,
    ('POST',   '/todos'): add,
    ('DELETE', '/todos'): clear,
}, port=8000) | run

Next up: websockets — broadcast + actor-per-connection. →