Embedding¶
The CLI (h2corn module:app) and the serve() function cover the common case: spawn the server as a top-level process. When you need finer control — running inside an existing event loop, supervising the server from your own code, or driving it from tests — reach for the Server class instead.
Inside an asyncio app¶
import asyncio
from h2corn import Config, Server
from hello import app
async def main() -> None:
server = Server(app, Config(bind=('127.0.0.1:8000',)))
await server.serve()
asyncio.run(main())
Server.serve() is an async function that runs until the server is asked to shut down. It is single-worker by design; when you need multiple workers, fall back to serve(), which goes through the same multi-process supervisor as the CLI.
Programmatic shutdown¶
Call shutdown() from any thread or coroutine to begin a graceful stop. In-flight requests get up to Config.timeout_graceful_shutdown seconds to complete.
import asyncio
from h2corn import Config, Server
from hello import app
async def main():
server = Server(app, Config(bind=('127.0.0.1:8000',)))
async def stop_after(delay: float):
await asyncio.sleep(delay)
server.shutdown()
await asyncio.gather(server.serve(), stop_after(5.0))
asyncio.run(main())
Binding to any free port¶
Bind port 0 and read the kernel-assigned address back from Server.addresses — ideal for test harnesses and service discovery:
server = Server(app, Config(bind=('127.0.0.1:0',)))
task = asyncio.create_task(server.serve())
while not server.addresses:
await asyncio.sleep(0)
print(server.addresses) # ('127.0.0.1:54123',)
When several TCP listeners all bind port 0 (for example 0.0.0.0:0 plus [::]:0), they deliberately share one kernel-assigned port.
Which entrypoint to use¶
| You want… | Use |
|---|---|
| The standard CLI experience, multi-worker | h2corn module:app |
| The same behavior from Python | h2corn.serve(app, config) |
| A single worker inside your own event loop | h2corn.Server(app, config).serve() |
| To embed in a test harness with programmatic stop | Server + shutdown() |