Add restate.serve: serve an app over HTTP/2

README.md

@@ -42,26 +42,23 @@ async def greet(ctx: restate.Context, name: str) -> str:
 app = restate.app(services=[greeter])
 ```
 
-`restate.app(...)` returns an ASGI application, so it can be served by any ASGI server.
-
-With [Hypercorn](https://github.com/pgjones/hypercorn):
-
-```shell
-pip install hypercorn
-python -m hypercorn app:app --bind "0.0.0.0:9080"
-```
-
-Or start it from code:
+Serve it with `restate.serve`, which needs the `serve` extra (`pip install "restate-sdk[serve]"`,
+Python 3.11 or later). It listens on `$PORT`, or 9080, with the HTTP/2 that Restate connects with, and
+stops on SIGINT or SIGTERM:
 
 ```python
 if __name__ == "__main__":
     import asyncio
-    import hypercorn
-    import hypercorn.asyncio
 
-    conf = hypercorn.Config()
-    conf.bind = ["0.0.0.0:9080"]
-    asyncio.run(hypercorn.asyncio.serve(app, conf))
+    asyncio.run(restate.serve(app))
+```
+
+`restate.app(...)` returns an ASGI application, so any ASGI server can serve it instead, for example
+[Hypercorn](https://github.com/pgjones/hypercorn), which also works on Python 3.10:
+
+```shell
+pip install hypercorn
+python -m hypercorn app:app --bind "0.0.0.0:9080"
 ```
 
 Once served, register the deployment with `restate deployments register http://localhost:9080`.

pyproject.toml

@@ -26,6 +26,10 @@ lint = ["mypy>=1.11.2", "pyright>=1.1.390", "ruff>=0.6.9"]
 harness = ["testcontainers", "hypercorn", "httpx"]
 serde = ["dacite", "pydantic", "msgspec"]
 client = ["httpx[http2]"]
+serve = [
+    "uvicorn>=0.53 ; python_version >= '3.11'",
+    "zttp>=0.0.34 ; python_version >= '3.11'",
+]
 adk = ["google-adk>=1.20.0"]
 openai = ["openai-agents>=0.6.1"]
 pydantic_ai = ["pydantic-ai-slim>=2.0,<3"]

python/restate/__init__.py

@@ -44,6 +44,7 @@
 
 from .endpoint import app
 from .connection import connect, with_connection, ConnectError
+from .serving import serve
 
 from .logging import getLogger, RestateLoggingFilter
 
@@ -114,6 +115,7 @@ async def create_client(
     "JournalValueCodecProvider",
     "app",
     "connect",
+    "serve",
     "with_connection",
     "ConnectError",
     "create_test_harness",

python/restate/serving.py

@@ -0,0 +1,92 @@
+#
+#  Copyright (c) 2023-2026 - Restate Software, Inc., Restate GmbH
+#
+#  This file is part of the Restate SDK for Python,
+#  which is released under the MIT license.
+#
+#  You can find a copy of the license in file LICENSE in the root
+#  directory of this repository or package, or at
+#  https://github.com/restatedev/sdk-typescript/blob/main/LICENSE
+#
+"""Serve a Restate app over HTTP/2: ``restate.serve``, the counterpart of ``restate.connect``."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import importlib
+import os
+import re
+import signal
+import typing
+
+from restate.server import install_shutdown_hook
+from restate.server_types import RestateAppT
+
+GRACEFUL_SHUTDOWN_SECONDS = 3
+"""How long a stopping server waits for the requests in flight; Restate retries the rest elsewhere."""
+
+_MISSING = (
+    "restate.serve runs on uvicorn's HTTP/2: install restate-sdk[serve] (uvicorn 0.53 or later, and zttp; "
+    "Python 3.11 or later), or serve the app with an ASGI server of your own, such as hypercorn"
+)
+
+
+async def serve(app: RestateAppT, *, host: str = "0.0.0.0", port: typing.Optional[int] = None) -> None:
+    """
+    Serve ``app`` over HTTP/2, until SIGINT or SIGTERM stops it.
+
+    Restate connects to a deployment with cleartext HTTP/2, which this serves (on uvicorn)::
+
+        app = restate.app([greeter])
+
+        if __name__ == "__main__":
+            asyncio.run(restate.serve(app))
+
+    It stops within a few seconds of SIGINT or SIGTERM: the invocations in flight are asked to wind
+    down, and Restate retries those that don't finish in time.
+
+    Args:
+        app: The app to serve, as ``restate.app`` returns it.
+        host: The address to listen on. Defaults to all interfaces.
+        port: The port to listen on (0 for one the OS picks). Defaults to ``$PORT``, else 9080.
+
+    Raises:
+        ImportError: If the ``serve`` extra is not installed (``pip install restate-sdk[serve]``).
+    """
+    try:
+        uvicorn = importlib.import_module("uvicorn")
+        importlib.import_module("zttp")
+    except ImportError as error:
+        raise ImportError(_MISSING) from error
+    version = re.match(r"(\d+)\.(\d+)", str(getattr(uvicorn, "__version__", "")))
+    if version is None or (int(version[1]), int(version[2])) < (0, 53):
+        raise ImportError(_MISSING)
+
+    config = uvicorn.Config(
+        app,
+        host=host,
+        port=port if port is not None else int(os.environ.get("PORT", "9080")),
+        http="zttp",
+        http2=True,
+        # The app has no lifespan, and a graceful shutdown would otherwise wait for Restate's open connections.
+        lifespan="off",
+        timeout_graceful_shutdown=GRACEFUL_SHUTDOWN_SECONDS,
+    )
+    server = uvicorn.Server(config)
+
+    # The app installs its own SIGTERM hook on its first request, through install_shutdown_hook, which keeps
+    # one asyncio handler per signal: it would replace uvicorn's handlers. So the server's exit goes through
+    # the same hook, registered first, and uvicorn captures no signals itself: both then run.
+    loop = asyncio.get_running_loop()
+    try:
+        for signum in (signal.SIGINT, signal.SIGTERM):
+            install_shutdown_hook(loop, _exit_on(server, signum), signum)
+        server.capture_signals = contextlib.nullcontext
+    except (NotImplementedError, RuntimeError, ValueError):
+        pass  # Windows, or not the main thread: uvicorn handles the signals itself, as it can
+    await server.serve()
+
+
+def _exit_on(server: typing.Any, signum: int) -> typing.Callable[[], None]:
+    return lambda: server.handle_exit(signum, None)

tests/serve.py

@@ -0,0 +1,116 @@
+#
+#  Copyright (c) 2023-2026 - Restate Software, Inc., Restate GmbH
+#
+#  This file is part of the Restate SDK for Python,
+#  which is released under the MIT license.
+#
+#  You can find a copy of the license in file LICENSE in the root
+#  directory of this repository or package, or at
+#  https://github.com/restatedev/sdk-typescript/blob/main/LICENSE
+#
+"""restate.serve: a program serves its app over HTTP/2, and stops on SIGTERM."""
+
+# pylint: disable=missing-function-docstring
+
+import asyncio
+import importlib.util
+import os
+import pathlib
+import signal
+import subprocess
+import sys
+import time
+
+import httpx
+import pytest
+
+from restate.harness import RestateContainer, find_free_port
+
+
+@pytest.fixture(scope="session")
+def anyio_backend():
+    return "asyncio"
+
+
+WITHOUT_UVICORN = """
+import asyncio, sys
+sys.modules["uvicorn"] = None
+import restate
+try:
+    asyncio.run(restate.serve(restate.app([])))
+except ImportError as error:
+    print(error)
+"""
+
+
+def test_without_the_serve_extra_it_says_what_to_install():
+    ran = subprocess.run([sys.executable, "-c", WITHOUT_UVICORN], capture_output=True, text=True, check=False)
+    assert ran.returncode == 0, ran.stderr
+    assert "install restate-sdk[serve]" in ran.stdout
+
+
+PROGRAM = """
+import asyncio
+import restate
+
+greeter = restate.Service("greeter")
+
+@greeter.handler()
+async def greet(ctx: restate.Context, name: str) -> str:
+    return f"Hello {name}!"
+
+asyncio.run(restate.serve(restate.app([greeter])))
+"""
+
+
+def serves_http2() -> bool:
+    return importlib.util.find_spec("uvicorn") is not None and importlib.util.find_spec("zttp") is not None
+
+
+@pytest.mark.anyio
+@pytest.mark.skipif(not serves_http2(), reason="needs the serve extra (Python 3.11+)")
+async def test_a_program_serves_its_app_over_http2_and_stops_on_sigterm(tmp_path: pathlib.Path):
+    # Restate first: Docker maps its ports to free host ports, which could otherwise take the program's.
+    container = RestateContainer("ghcr.io/restatedev/restate:latest", always_replay=False, disable_retries=False)
+    container.start()
+    port = find_free_port()
+    output = tmp_path / "program.log"
+    with output.open("wb") as log:
+        program = subprocess.Popen(
+            [sys.executable, "-c", PROGRAM], env={**os.environ, "PORT": str(port)}, stdout=log, stderr=log
+        )
+    try:
+        await listening(program, port, output)
+        async with httpx.AsyncClient(base_url=container.admin_url(), timeout=30) as admin:
+            uri = f"http://host.docker.internal:{port}"
+            registered = await admin.post("/deployments", json={"uri": uri, "force": True})
+            assert registered.status_code < 300, registered.text
+            deployment = (await admin.get(f"/deployments/{registered.json()['id']}")).json()
+            assert deployment.get("http_version") == "HTTP/2.0", deployment
+        async with httpx.AsyncClient(timeout=30) as ingress:
+            answer = await ingress.post(f"{container.ingress_url()}/greeter/greet", json="Alice")
+        assert answer.json() == "Hello Alice!"
+
+        # Stopped while Restate still holds its connections open: it stops within its graceful timeout.
+        program.send_signal(signal.SIGTERM)
+        started = time.monotonic()
+        program.wait(timeout=10)
+        assert time.monotonic() - started < 10, output.read_text()
+    finally:
+        program.kill()
+        container.stop()
+
+
+async def listening(program: subprocess.Popen[bytes], port: int, output: pathlib.Path) -> None:
+    """Wait until ``program`` accepts connections on ``port``, or fail with what it printed."""
+    for _ in range(200):
+        assert program.poll() is None, f"the program exited:\n{output.read_text()}"
+        try:
+            _, writer = await asyncio.open_connection("127.0.0.1", port)
+        except OSError:
+            await asyncio.sleep(0.05)
+            continue
+        writer.close()
+        await writer.wait_closed()
+        return
+    raise AssertionError(f"the program never listened on {port}:\n{output.read_text()}")