Async Integration

async def serve(
    app: ASGIFramework | WSGIFramework,
    config: Config,
    *,
    shutdown_trigger: Callable[..., Awaitable[None]] | None = None,
    mode: str | None = None
) -> None:
    """
    Programmatic interface for serving applications with asyncio.
    
    Starts a Hypercorn server within an asyncio event loop, providing
    full control over server lifecycle and shutdown handling. Ideal for
    embedding Hypercorn within larger asyncio applications or for
    custom server management.
    
    Args:
        app: ASGI or WSGI application to serve
        config: Config object with server settings
        shutdown_trigger: Optional async callable that when awaited
                         triggers graceful server shutdown
        mode: Optional mode specification ("asgi" or "wsgi").
              If not provided, mode is auto-detected
              
    Returns:
        None - function completes when server shuts down
        
    Example shutdown triggers:
        - asyncio.Event.wait()
        - signal handling coroutines
        - custom application shutdown logic
    """

async def serve(
    app: ASGIFramework | WSGIFramework, 
    config: Config,
    *,
    shutdown_trigger: Callable[..., Awaitable[None]] | None = None,
    task_status: trio._core._run._TaskStatus = trio.TASK_STATUS_IGNORED,
    mode: str | None = None
) -> None:
    """
    Programmatic interface for serving applications with trio.
    
    Starts a Hypercorn server within a trio nursery, with support for
    trio's structured concurrency patterns and task status reporting.
    
    Args:
        app: ASGI or WSGI application to serve
        config: Config object with server settings  
        shutdown_trigger: Optional async callable that when awaited
                         triggers graceful server shutdown
        task_status: Trio task status object for nursery integration
        mode: Optional mode specification ("asgi" or "wsgi").
              If not provided, mode is auto-detected
              
    Returns:
        None - function completes when server shuts down
        
    Task status is used with trio nurseries to indicate when
    the server has started and is ready to accept connections.
    """

# ASGI application type
ASGIFramework = Callable[[Scope, ASGIReceiveCallable, ASGISendCallable], Awaitable[None]]

# WSGI application type  
WSGIFramework = Callable[[dict, Callable], Iterable[bytes]]

# Union type for framework detection
Framework = ASGIFramework | WSGIFramework

import asyncio
from hypercorn.config import Config
from hypercorn.asyncio import serve

async def app(scope, receive, send):
    """Simple ASGI application"""
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/plain']],
    })
    await send({
        'type': 'http.response.body', 
        'body': b'Hello World',
    })

async def main():
    config = Config()
    config.bind = ["127.0.0.1:8000"]
    
    # Serve until manually stopped
    await serve(app, config)

asyncio.run(main())

import asyncio
import signal
from hypercorn.config import Config
from hypercorn.asyncio import serve

async def shutdown_trigger():
    """Wait for SIGTERM or SIGINT"""
    shutdown_event = asyncio.Event()
    
    def signal_handler():
        shutdown_event.set()
        
    loop = asyncio.get_event_loop()
    loop.add_signal_handler(signal.SIGTERM, signal_handler)
    loop.add_signal_handler(signal.SIGINT, signal_handler)
    
    await shutdown_event.wait()

async def main():
    config = Config()
    config.bind = ["0.0.0.0:8000"]
    
    # Serve with graceful shutdown on signals
    await serve(app, config, shutdown_trigger=shutdown_trigger)

asyncio.run(main())

import trio
from hypercorn.config import Config
from hypercorn.trio import serve

async def app(scope, receive, send):
    """Simple ASGI application"""
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [[b'content-type', b'text/plain']],
    })
    await send({
        'type': 'http.response.body',
        'body': b'Hello World',
    })

async def main():
    config = Config()  
    config.bind = ["127.0.0.1:8000"]
    
    async with trio.open_nursery() as nursery:
        # Start server in nursery
        nursery.start_soon(serve, app, config, task_status=trio.TASK_STATUS_IGNORED)
        
        # Server is now running - do other work
        await trio.sleep(60)  # Run for 60 seconds
        
        # Nursery cleanup will shutdown server

trio.run(main)

import asyncio
from hypercorn.config import Config
from hypercorn.asyncio import serve

class ServerManager:
    def __init__(self):
        self.shutdown_event = asyncio.Event()
        
    async def start_server(self, app):
        config = Config()
        config.bind = ["0.0.0.0:8000"]
        
        # Serve with custom shutdown trigger
        await serve(app, config, shutdown_trigger=self.shutdown_event.wait)
        
    def shutdown(self):
        """Trigger graceful shutdown"""
        self.shutdown_event.set()

# Usage
manager = ServerManager()

# In one task
asyncio.create_task(manager.start_server(app))

# Later, from another part of application
manager.shutdown()  # Triggers graceful shutdown

import asyncio
from hypercorn.config import Config
from hypercorn.asyncio import serve

def wsgi_app(environ, start_response):
    """Simple WSGI application"""
    status = '200 OK'
    headers = [('Content-type', 'text/plain')]
    start_response(status, headers)
    return [b'Hello World']

async def main():
    config = Config()
    config.bind = ["127.0.0.1:8000"]
    
    # Serve WSGI app (mode auto-detected)
    await serve(wsgi_app, config)

asyncio.run(main())

async def main():
    config = Config()
    config.bind = ["127.0.0.1:8000"]
    
    # Explicitly specify mode (useful for ambiguous applications)
    await serve(app, config, mode="asgi")
    # or
    await serve(wsgi_app, config, mode="wsgi")