Server: start_lambda_server

This page documents start_lambda_server — the function that configures and starts the FastAPI/uvicorn HTTP server for your lambda service.

start_lambda_server(service, *, logger=None, custom_args=None, run_opts=None, with_telemetry=None)

A helper function to start a FastAPI server for a lambda-style IVCAP service.

Parameters:

Name	Type	Description	Default
service	Service	service description	required
logger	Logger	optional logger instance; defaults to getLogger("app")	None
custom_args	Optional[Callable[[ArgumentParser], Namespace]]	callback to add extra CLI arguments. Defaults to None.	None
run_opts	Optional[Dict[str, Any]]	extra options forwarded to uvicorn.Config. Defaults to None.	None
with_telemetry	bool | None	(Optional[bool]): Instantiate or block use of OpenTelemetry tracing	None

Source code in ivcap_lambda/server.py

def start_lambda_server(
    service: Service,
    *,
    logger: Logger | None = None,
    custom_args: Callable[[argparse.ArgumentParser], argparse.Namespace] | None = None,
    run_opts: dict[str, Any] | None = None,
    with_telemetry: bool | None = None,
):
    """A helper function to start a FastAPI server for a lambda-style IVCAP service.

    Args:
        service (Service): service description
        logger (Logger): optional logger instance; defaults to getLogger("app")
        custom_args (Optional[Callable[[argparse.ArgumentParser], argparse.Namespace]], optional): callback to add extra CLI arguments. Defaults to None.
        run_opts (Optional[Dict[str, Any]], optional): extra options forwarded to uvicorn.Config. Defaults to None.
        with_telemetry: (Optional[bool]): Instantiate or block use of OpenTelemetry tracing
    """
    if len(tools) == 0:
        raise ValueError(
            "No tools have been registered. Please register at least one tool using the ivcap_lambda decorator."
        )

    app = get_fast_app()
    app.title = service.name
    app.version = service.version or os.environ.get("VERSION", "???")
    app.contact = dict(service.contact) if service.contact else None
    app.license_info = dict(service.license) if service.license else None

    title = service.name
    if logger is None:
        logger = getLogger("app")

    tool_names = [tool.name for tool in tools]
    parser = argparse.ArgumentParser(description=title)
    parser.add_argument(
        "--host",
        type=str,
        default=os.environ.get("HOST", "0.0.0.0"),
        help="Host address",
    )
    parser.add_argument(
        "--port", type=int, default=os.environ.get("PORT", "8090"), help="Port number"
    )
    parser.add_argument(
        "--with-telemetry", action="store_true", help="Initialise OpenTelemetry"
    )
    parser.add_argument("--with-mcp", action="store_true", help="Add an MCP endpoint")
    parser.add_argument(
        "--with-mcp-stdio",
        action="store_true",
        help=(
            "Run as an MCP server over stdio instead of starting the HTTP "
            "server (for local development with stdio-based MCP clients, "
            "e.g. Claude Desktop, Cline). Requires the 'mcp' extra."
        ),
    )
    parser.add_argument(
        "--list-services",
        action="store_true",
        help="List the names of all registered services/tools and exit",
    )
    parser.add_argument(
        "--print-service-description",
        type=str,
        metavar="NAME",
        nargs="?",
        const=tool_names[0],
        default=None,
        help=f"Print service description to stdout [{','.join(tool_names)}]",
    )
    parser.add_argument(
        "--service-name",
        type=str,
        default=None,
        help=(
            "Override the service name used in --print-service-description "
            "(defaults to the tool name, with '_' replaced by '-')"
        ),
    )
    parser.add_argument(
        "--print-tool-description",
        type=str,
        metavar="NAME",
        nargs="?",
        const=tool_names[0],
        default=None,
        help=f"Print tool description to stdout [{','.join(tool_names)}]",
    )

    if custom_args is not None:
        args = custom_args(parser)
    else:
        args = parser.parse_args()

    if args.with_mcp_stdio and args.with_mcp:
        parser.error("--with-mcp-stdio and --with-mcp are mutually exclusive")

    if args.list_services:
        for name in tool_names:
            print(name)
        sys.exit(0)

    if args.print_tool_description:
        tool = next((t for t in tools if t.name == args.print_tool_description), None)
        if tool is None:
            print(
                f"Tool '{args.print_tool_description}' not found. Available tools: {', '.join(tool_names)}",
                file=sys.stderr,
            )
            sys.exit(1)
        print_tool_definition(tool.worker_fn)
        sys.exit(0)

    if args.print_service_description:
        from .service_definition import print_rest_service_definition

        tool = next(
            (t for t in tools if t.name == args.print_service_description), None
        )
        if tool is None:
            print(
                f"Tool '{args.print_service_description}' not found. Available tools: {', '.join(tool_names)}",
                file=sys.stderr,
            )
            sys.exit(1)
        tool_slug = tool.name.replace("_", "-")
        name = args.service_name or f"{service.name}-{tool_slug}"
        print_rest_service_definition(
            service, tool.worker_fn, name=name, path=tool.path_prefix
        )
        sys.exit(0)

    logger.info(
        f"{title} - {os.getenv('VERSION')} - v{get_version()}|v{get_service_version()}"
    )

    # print(f">>>> OTEL_EXPORTER_OTLP_ENDPOINT: {os.environ.get('OTEL_EXPORTER_OTLP_ENDPOINT')}")
    set_event_reporter_factory(SidecarReporter)  # type: ignore[arg-type]

    def get_context():
        jctxt = get_job_context()
        if jctxt is None or jctxt.job_id is None:
            logger.warning("missing job context in thread")
            return None
        return jctxt

    set_context(get_context)

    if args.with_mcp_stdio:
        # Run as a stdio MCP server instead of the HTTP/uvicorn server: no
        # FastAPI app involved, so there is nothing to instrument there, but
        # outbound httpx/requests calls (e.g. via JobContext.ivcap) are still
        # instrumented the same way as for the HTTP server.
        from .mcp import run_mcp_stdio

        otel_instrument(with_telemetry, None, logger)
        run_mcp_stdio(service.name, app.version)
        return

    # Check for '_healtz' service
    healtz = find_first(app.routes, lambda r: r.path == "/_healtz")
    if healtz is None:

        @app.get("/_healtz", tags=["System"])
        def healtz():
            return {"version": os.environ.get("VERSION", "???")}

    if args.with_mcp:
        from .mcp import register_mcp

        register_mcp(app, "/mcp")

    otel_instrument(
        with_telemetry, lambda _: FastAPIInstrumentor.instrument_app(app), logger
    )

    async def _add_version(request: Request, call_next) -> Response:
        from .version import __version__

        resp = await call_next(request)
        resp.headers["Ivcap-AI-Tool-Version"] = __version__
        return resp

    app.middleware("http")(_add_version)

    if run_opts is None:
        run_opts = {}

    class Server(uvicorn.Server):
        def handle_exit(self, sig: int, frame: Any) -> None:
            logger.info(  # type: ignore[union-attr]
                "Received request for shutdown. Waiting for all running requests to finish first."
            )
            Executor.wait_for_exit_ready()
            super().handle_exit(sig, frame)

    server = Server(
        config=uvicorn.Config(
            app,
            host=args.host,
            port=args.port,
            log_config=service_log_config(),
            **run_opts,
        )
    )

    # Start the server
    server.run()

---

get_fast_app

Returns the shared FastAPI application instance. Useful when you need to add custom routes or middleware:

get_fast_app()

Get the FastAPI app instance.

Returns:

Name	Type	Description
FastAPI	FastAPI	The FastAPI app instance.

Source code in ivcap_lambda/server.py

def get_fast_app() -> FastAPI:
    """Get the FastAPI app instance.

    Returns:
        FastAPI: The FastAPI app instance.
    """
    return _app

---

Deprecated: start_tool_server

start_tool_server(*args, **kwargs)

Deprecated alias for :func:start_lambda_server.

.. deprecated:: Use :func:start_lambda_server instead.

Source code in ivcap_lambda/server.py

def start_tool_server(*args, **kwargs):
    """Deprecated alias for :func:`start_lambda_server`.

    .. deprecated::
        Use :func:`start_lambda_server` instead.
    """
    import warnings

    warnings.warn(
        "start_tool_server() is deprecated; use start_lambda_server() instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return start_lambda_server(*args, **kwargs)

Copy Markdown