Skip to content

Python endpoints (@peach.endpoint)

@peach.endpoint turns a plain Python function into an HTTP endpoint — a Ray Serve deployment — without writing a @serve.deployment class, and without declaring anything in peach.conf.

from pipe_algorithms_lib.compute import peach

@peach.endpoint(name="episodes", py_env="ml")
def episodes(u: str, size: int = 30):
    return [{"id": "a"}, {"id": "b"}]

The decorator itself only describes the endpoint. There are two ways to actually run it:

How Lives for
Production put the file in your codops repo's peach/ folder; the CI deploys it until you change it
Prototyping call peach_serve(fn) from a notebook until the next ArgoCD sync, or its ttl

Both run the same function with the same request handling and the same response shape, so a prototype's callers see the shape they will keep once the code moves into production.

@peach.task — the other half of the peach namespace — is covered on the Peach decorators page.

Writing the function

Two signature shapes are accepted, and the decorator picks between them automatically.

Typed parameters (the usual choice)

Annotate your parameters and they are bound from the HTTP request and coerced to their annotations, FastAPI-style:

@peach.endpoint(name="episodes", route_prefix="/episodes", py_env="ml")
def episodes(
    u: str,                             # required — 422 if missing
    size: int = 30,                     # ?size=5   -> 5, not "5"
    limit_episodes: int | None = None,  # omitted   -> None
    with_duration: bool = False,        # ?with_duration=true -> True
    platform: str = "urplay",
):
    return {"u": u, "size": size}

Scalars come from the query string; annotate a parameter with a pydantic model or a dict to read it from a JSON body instead. GET and POST are both routed, and the endpoint answers with or without a trailing slash. A parameter that does not parse gets a 422 naming the offending field, rather than a 500 from deep inside the replica.

Calling the function directly in Python still works normally — episodes(u="x", size=5).

A raw request

A single unannotated parameter (or one annotated Request) gets the raw Starlette request and does its own parsing:

@peach.endpoint(name="hello-demo")
def hello(request):
    return {"hello": request.query_params.get("name", "world")}

A lone unannotated parameter is a Request

def hello(u) is read as this shape and receives a Request, not a query parameter named u. Annotate it — def hello(u: str) — to get the binding.

Either shape may be async.

Loading a model once, with init_fn

init_fn is a zero-argument callable run exactly once, when the deployment replica starts — the same timing as a class-based deployment's __init__, not at decoration time and not per request. Its return value is injected as a module-level global named init_state, which your function reads directly:

def init_model():
    return {"cf": load_cf_model(CF_MODEL_BASE)}

@peach.endpoint(name="reco", init_fn=init_model)
def reco(u: str, size: int = 30):
    return init_state["cf"].recommend(u, size)

init_fn is also the answer to this error on the decorator line:

Could not serialize the deployment ... cannot pickle '_thread.lock' object

Applying the decorator pickles your function, and that captures the globals it reads — so a module-level redis client, model handle or database connection (anything holding a lock or a socket) fails there. Build it in init_fn and read it off init_state, so it is constructed on the replica and never crosses the wire.

For anything heavier — constructor arguments, several pieces of state — write a @serve.deployment class instead.

The response envelope

By default your return value is wrapped in the same envelope a production Peach endpoint uses:

@peach.endpoint(name="reco", route_prefix="/reco")
def reco(u: str, size: int = 30):
    return [{"id": "a"}, {"id": "b"}]

GET /reco?u=x&size=2

{"result": {"items": [{"id": "a"}, {"id": "b"}],
            "id": "<recsys token>",
            "fallback_used": "false"},
 "status": "ok"}

What the envelope does with what you return:

You return Becomes
[a, b] items: [a, b]
{"items": [a, b]} items: [a, b] (unwrapped one level)
{"foo": 1} items: {"foo": 1} (a bare dict is the payload)
([a, b], {"total": 9}) items: [a, b], total: 9

A 2-tuple is read as (items, meta) and meta is spread next to items — so meta can overwrite items, but id is written last and can never be overwritten. A headers key in meta is promoted to HTTP headers and stays in the body too, as in production.

Pass wrap_response=False to get your return value as the response body instead, unwrapped.

Debugging a request

Adding ?debug=true to the request adds the production debug block:

"debug": {"algorithm_used": {"method": "reco", "parameters": {"u": "x"}},
          "execution_ms": 12.4,
          "version": "dev"}

Errors

Raising returns a 500 with {"status": "error", "message": ...}, where the message is generic. ?debug=true returns str(exc) and the traceback instead.

The detail is withheld by default on purpose: these apps are published on the shared api.{codops} ingress, so the caller is not necessarily the author, and an exception's own text routinely carries a connection string or a host name. The full traceback always goes to the replica log either way.

Two production behaviours this does not reproduce

fallback_used is always "false" — there is no fallback component to route to, so an empty items list stays a 200 rather than triggering one. And an upstream 429/503 is not relayed as back-pressure. Both belong to the codops gateway rather than to a single component.

Deploying to production

Put the decorated function in a peach/ folder at the top level of your codops repo:

peach/
    codops.txt      — a single line naming your codops id, e.g. "sesr"
    endpoints.py    — top-level .py files only, not recursive

The CI discovers every @peach.endpoint in those files by static analysis (it never imports them), generates a standalone Ray Serve application per endpoint, and adds it to the codops' Serve config. Each one is published at:

https://api.<codops>.<envDomain>/v2<route_prefix or /name>

Any live peach_serve(...), peach_serve_status() or peach_serve_delete(...) calls in the file are stripped from the generated code, so you can leave your prototyping lines in place (commented or not) without them running at import time.

@peach.task functions in the same folder are picked up too — with cron= they become scheduled Prefect flows, without it they are just helpers your endpoint can call. See Running a task on a schedule.

Prototyping from a notebook

peach_serve() runs the deployment live over the same ray:// connection peach.task uses, so you can hit a real HTTP endpoint from a notebook in seconds.

Ephemeral / dev-only — not a deployment mechanism

peach_serve() is not wired into your codops' serveConfigV2 / ArgoCD-managed Serve config, and will disappear the next time KubeRay reconciles that config against the real cluster state. For production, use the peach/ folder.

from pipe_algorithms_lib.compute import (
    peach, peach_serve, peach_serve_status, peach_serve_delete,
)

@peach.endpoint(name="hello-demo", py_env="base", ttl=3600)
def hello(request):
    return {"message": "hello from ray serve"}

handle = peach_serve(hello)
peach_serve_status()
peach_serve_delete("hello-demo")

peach_serve() prints where the app landed:

Ray Serve app 'tmp_peach-serve-hello-demo' running (EPHEMERAL / DEV-ONLY — not production)
  api: https://api.<codops>.peach.ebu.io/tmp/hello-demo
  dashboard: https://ray.<codops>.peach.ebu.io/#/serve

Ephemeral apps always live under /tmp/, so they cannot collide with the real production app's /v2 route. route_prefix is a stand-in for the name within that namespace, not a full path.

Cleaning up

  • ttl=<seconds> on the decorator (or on peach_serve()) auto-deletes the app after that long, so you do not have to remember. Redeploying the same name cancels the pending timer and starts a fresh one.
  • peach_serve_status() lists what is running, flagging which apps are peach_serve-managed and which are not.
  • peach_serve_delete(name) tears down exactly that one app.

Never call serve.shutdown()

It would tear down the entire Serve instance for your codops, including the real production app. Use peach_serve_delete(name).

Class-based deployments

For stateful deployments, write the @serve.deployment class exactly as you would for a real peach-jobs deployment and pass it — with its constructor arguments — to peach_serve():

from ray import serve

@serve.deployment
class MyModel:
    def __init__(self, greeting="Hello"):
        self.greeting = greeting

    async def __call__(self, request):
        return {"message": self.greeting}

handle = peach_serve(MyModel, "Hello", name="my-demo", py_env="base")

name is required here, since there is no decorator to read it from.

Note

@peach.endpoint functions take no constructor arguments — passing bind args to peach_serve() for one raises a ValueError. Use init_fn, or a class.

Calling through the handle

peach_serve() returns the Serve DeploymentHandle. Calling through it invokes the deployment directly as a Python call, bypassing HTTP — so a raw-request handler needs a compatible object passed in. Typed-parameter endpoints are served through an ASGI app, so the handle addresses that app rather than your function: drive those over HTTP, and use a @serve.deployment class if you need handle.remote() calls.

Blocking calls in a raw-request handler

A synchronous raw-request function runs on the replica's asyncio event loop, so a blocking call inside it (a peach.task's own ray.get(), for example) freezes the whole replica until it returns. Typed-parameter handlers do not have this problem — FastAPI runs sync ones in a threadpool — and neither do production deployments generated from peach/.

Arguments

Argument Meaning
name identifies the endpoint. Required. Also the default route.
route_prefix the path to serve at, under /v2 in production and /tmp/ when prototyping. Defaults to /<name>.
py_env the Python environment to run in. Defaults to base.
packages ad-hoc extra packages; same caveats as on peach.task.
ray_actor_options resource overrides for the replica (num_cpus, num_gpus, memory, ...).
init_fn zero-argument callable run once per replica; result exposed as init_state.
ttl prototyping only — auto-delete the ephemeral app after this many seconds.
wrap_response wrap the return value in the production envelope. True by default.

peach_serve() accepts name, route_prefix, py_env, packages, ray_actor_options and ttl too, and anything passed there overrides what the decorator captured.