api_server.py 51.5 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
import asyncio
4
import hashlib
5
6
import importlib
import inspect
7
import json
8
import multiprocessing
9
import multiprocessing.forkserver as forkserver
10
import os
11
import secrets
12
import signal
13
import socket
14
import tempfile
15
import uuid
16
from argparse import Namespace
17
from collections.abc import AsyncGenerator, AsyncIterator, Awaitable
18
from contextlib import asynccontextmanager
19
from http import HTTPStatus
20
from typing import Annotated, Any, Literal
21

22
import model_hosting_container_standards.sagemaker as sagemaker_standards
23
import pydantic
24
import uvloop
25
from fastapi import APIRouter, Depends, FastAPI, Form, HTTPException, Query, Request
Zhuohan Li's avatar
Zhuohan Li committed
26
27
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
28
from fastapi.responses import JSONResponse, Response, StreamingResponse
29
from starlette.concurrency import iterate_in_threadpool
30
31
from starlette.datastructures import URL, Headers, MutableHeaders, State
from starlette.types import ASGIApp, Message, Receive, Scope, Send
Zhuohan Li's avatar
Zhuohan Li committed
32

33
import vllm.envs as envs
34
from vllm.config import VllmConfig
Woosuk Kwon's avatar
Woosuk Kwon committed
35
from vllm.engine.arg_utils import AsyncEngineArgs
36
from vllm.engine.protocol import EngineClient
37
38
39
40
41
42
43
from vllm.entrypoints.anthropic.protocol import (
    AnthropicError,
    AnthropicErrorResponse,
    AnthropicMessagesRequest,
    AnthropicMessagesResponse,
)
from vllm.entrypoints.anthropic.serving_messages import AnthropicServingMessages
44
from vllm.entrypoints.launcher import serve_http
45
from vllm.entrypoints.logger import RequestLogger
46
from vllm.entrypoints.openai.cli_args import make_arg_parser, validate_parsed_serve_args
47
from vllm.entrypoints.openai.orca_metrics import metrics_header
48
49
50
51
52
53
54
55
56
57
58
from vllm.entrypoints.openai.protocol import (
    ChatCompletionRequest,
    ChatCompletionResponse,
    CompletionRequest,
    CompletionResponse,
    ErrorInfo,
    ErrorResponse,
    ResponsesRequest,
    ResponsesResponse,
    StreamingResponsesResponse,
    TranscriptionRequest,
59
    TranscriptionResponseVariant,
60
    TranslationRequest,
61
    TranslationResponseVariant,
62
)
63
64
from vllm.entrypoints.openai.serving_chat import OpenAIServingChat
from vllm.entrypoints.openai.serving_completion import OpenAIServingCompletion
65
from vllm.entrypoints.openai.serving_engine import OpenAIServing
66
67
68
69
from vllm.entrypoints.openai.serving_models import (
    BaseModelPath,
    OpenAIServingModels,
)
70
from vllm.entrypoints.openai.serving_responses import OpenAIServingResponses
71
from vllm.entrypoints.openai.serving_transcription import (
72
73
74
    OpenAIServingTranscription,
    OpenAIServingTranslation,
)
75
from vllm.entrypoints.openai.tool_parsers import ToolParserManager
76
77
78
79
80
from vllm.entrypoints.openai.utils import validate_json_request
from vllm.entrypoints.pooling.classify.serving import ServingClassification
from vllm.entrypoints.pooling.embed.serving import OpenAIServingEmbedding
from vllm.entrypoints.pooling.pooling.serving import OpenAIServingPooling
from vllm.entrypoints.pooling.score.serving import ServingScores
81
82
83
84
85
from vllm.entrypoints.serve.disagg.serving import ServingTokens
from vllm.entrypoints.serve.elastic_ep.middleware import (
    ScalingMiddleware,
)
from vllm.entrypoints.serve.tokenize.serving import OpenAIServingTokenization
86
87
88
89
90
from vllm.entrypoints.tool_server import DemoToolServer, MCPToolServer, ToolServer
from vllm.entrypoints.utils import (
    cli_env_setup,
    load_aware_call,
    log_non_default_args,
91
92
    process_chat_template,
    process_lora_modules,
93
94
    with_cancellation,
)
95
from vllm.logger import init_logger
96
from vllm.reasoning import ReasoningParserManager
97
from vllm.tasks import POOLING_TASKS
yhu422's avatar
yhu422 committed
98
from vllm.usage.usage_lib import UsageContext
Cyrus Leung's avatar
Cyrus Leung committed
99
from vllm.utils.argparse_utils import FlexibleArgumentParser
100
from vllm.utils.gc_utils import freeze_gc_heap
101
from vllm.utils.network_utils import is_valid_ipv6_address
102
from vllm.utils.system_utils import decorate_logs, set_ulimit
103
from vllm.version import __version__ as VLLM_VERSION
Zhuohan Li's avatar
Zhuohan Li committed
104

105
prometheus_multiproc_dir: tempfile.TemporaryDirectory
106

107
# Cannot use __name__ (https://github.com/vllm-project/vllm/pull/4765)
108
logger = init_logger("vllm.entrypoints.openai.api_server")
109

110
111
ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL = "endpoint-load-metrics-format"

112
_running_tasks: set[asyncio.Task] = set()
113

114

115
@asynccontextmanager
116
async def lifespan(app: FastAPI):
117
118
    try:
        if app.state.log_stats:
119
            engine_client: EngineClient = app.state.engine_client
120
121
122

            async def _force_log():
                while True:
123
                    await asyncio.sleep(envs.VLLM_LOG_STATS_INTERVAL)
124
                    await engine_client.do_log_stats()
125
126
127
128
129
130

            task = asyncio.create_task(_force_log())
            _running_tasks.add(task)
            task.add_done_callback(_running_tasks.remove)
        else:
            task = None
131
132
133

        # Mark the startup heap as static so that it's ignored by GC.
        # Reduces pause times of oldest generation collections.
134
        freeze_gc_heap()
135
136
137
138
139
140
141
142
        try:
            yield
        finally:
            if task is not None:
                task.cancel()
    finally:
        # Ensure app state including engine ref is gc'd
        del app.state
143
144


145
@asynccontextmanager
146
async def build_async_engine_client(
147
    args: Namespace,
148
149
    *,
    usage_context: UsageContext = UsageContext.OPENAI_API_SERVER,
150
151
    disable_frontend_multiprocessing: bool | None = None,
    client_config: dict[str, Any] | None = None,
152
) -> AsyncIterator[EngineClient]:
153
154
155
156
    if os.getenv("VLLM_WORKER_MULTIPROC_METHOD") == "forkserver":
        # The executor is expected to be mp.
        # Pre-import heavy modules in the forkserver process
        logger.debug("Setup forkserver with pre-imports")
157
        multiprocessing.set_start_method("forkserver")
158
159
160
161
        multiprocessing.set_forkserver_preload(["vllm.v1.engine.async_llm"])
        forkserver.ensure_running()
        logger.debug("Forkserver setup complete!")

162
    # Context manager to handle engine_client lifecycle
163
164
    # Ensures everything is shutdown and cleaned up on error/exit
    engine_args = AsyncEngineArgs.from_cli_args(args)
165
166
167
    if client_config:
        engine_args._api_process_count = client_config.get("client_count", 1)
        engine_args._api_process_rank = client_config.get("client_index", 0)
168

169
    if disable_frontend_multiprocessing is None:
170
        disable_frontend_multiprocessing = bool(args.disable_frontend_multiprocessing)
171

172
    async with build_async_engine_client_from_engine_args(
173
174
175
176
        engine_args,
        usage_context=usage_context,
        disable_frontend_multiprocessing=disable_frontend_multiprocessing,
        client_config=client_config,
177
    ) as engine:
178
179
180
181
182
183
        yield engine


@asynccontextmanager
async def build_async_engine_client_from_engine_args(
    engine_args: AsyncEngineArgs,
184
185
    *,
    usage_context: UsageContext = UsageContext.OPENAI_API_SERVER,
186
    disable_frontend_multiprocessing: bool = False,
187
    client_config: dict[str, Any] | None = None,
188
) -> AsyncIterator[EngineClient]:
189
    """
190
    Create EngineClient, either:
191
192
193
194
195
196
        - in-process using the AsyncLLMEngine Directly
        - multiprocess using AsyncLLMEngine RPC

    Returns the Client or None if the creation failed.
    """

197
198
199
    # Create the EngineConfig (determines if we can use V1).
    vllm_config = engine_args.create_engine_config(usage_context=usage_context)

200
    if disable_frontend_multiprocessing:
201
        logger.warning("V1 is enabled, but got --disable-frontend-multiprocessing.")
202

203
    from vllm.v1.engine.async_llm import AsyncLLM
204

205
    async_llm: AsyncLLM | None = None
206
207
208
209
210
211

    # Don't mutate the input client_config
    client_config = dict(client_config) if client_config else {}
    client_count = client_config.pop("client_count", 1)
    client_index = client_config.pop("client_index", 0)

212
213
214
215
216
    try:
        async_llm = AsyncLLM.from_vllm_config(
            vllm_config=vllm_config,
            usage_context=usage_context,
            enable_log_requests=engine_args.enable_log_requests,
217
            aggregate_engine_logging=engine_args.aggregate_engine_logging,
218
219
220
            disable_log_stats=engine_args.disable_log_stats,
            client_addresses=client_config,
            client_count=client_count,
221
222
            client_index=client_index,
        )
223
224

        # Don't keep the dummy data in memory
225
        assert async_llm is not None
226
227
228
229
230
231
        await async_llm.reset_mm_cache()

        yield async_llm
    finally:
        if async_llm:
            async_llm.shutdown()
232
233


Ethan Xu's avatar
Ethan Xu committed
234
router = APIRouter()
Zhuohan Li's avatar
Zhuohan Li committed
235

236

237
238
239
240
241
def base(request: Request) -> OpenAIServing:
    # Reuse the existing instance
    return tokenization(request)


242
243
244
245
def models(request: Request) -> OpenAIServingModels:
    return request.app.state.openai_serving_models


246
def responses(request: Request) -> OpenAIServingResponses | None:
247
248
249
    return request.app.state.openai_serving_responses


250
251
252
253
def messages(request: Request) -> AnthropicServingMessages:
    return request.app.state.anthropic_serving_messages


254
def chat(request: Request) -> OpenAIServingChat | None:
255
256
257
    return request.app.state.openai_serving_chat


258
def completion(request: Request) -> OpenAIServingCompletion | None:
259
260
261
    return request.app.state.openai_serving_completion


262
263
def tokenization(request: Request) -> OpenAIServingTokenization:
    return request.app.state.openai_serving_tokenization
264
265


266
267
268
269
def transcription(request: Request) -> OpenAIServingTranscription:
    return request.app.state.openai_serving_transcription


270
271
272
273
def translation(request: Request) -> OpenAIServingTranslation:
    return request.app.state.openai_serving_translation


274
def engine_client(request: Request) -> EngineClient:
275
276
277
    return request.app.state.engine_client


278
279
280
281
def generate_tokens(request: Request) -> ServingTokens | None:
    return request.app.state.serving_tokens


282
283
284
285
286
287
288
@router.get("/load")
async def get_server_load_metrics(request: Request):
    # This endpoint returns the current server load metrics.
    # It tracks requests utilizing the GPU from the following routes:
    # - /v1/chat/completions
    # - /v1/completions
    # - /v1/audio/transcriptions
289
    # - /v1/audio/translations
290
291
    # - /v1/embeddings
    # - /pooling
292
    # - /classify
293
294
295
296
297
    # - /score
    # - /v1/score
    # - /rerank
    # - /v1/rerank
    # - /v2/rerank
298
    return JSONResponse(content={"server_load": request.app.state.server_load_metrics})
299
300


Ethan Xu's avatar
Ethan Xu committed
301
@router.get("/v1/models")
302
async def show_available_models(raw_request: Request):
303
    handler = models(raw_request)
304

305
306
    models_ = await handler.show_available_models()
    return JSONResponse(content=models_.model_dump())
Zhuohan Li's avatar
Zhuohan Li committed
307
308


Ethan Xu's avatar
Ethan Xu committed
309
@router.get("/version")
310
async def show_version():
311
    ver = {"version": VLLM_VERSION}
312
313
314
    return JSONResponse(content=ver)


315
async def _convert_stream_to_sse_events(
316
    generator: AsyncGenerator[StreamingResponsesResponse, None],
317
) -> AsyncGenerator[str, None]:
318
319
    """Convert the generator to a stream of events in SSE format"""
    async for event in generator:
320
        event_type = getattr(event, "type", "unknown")
321
        # https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format
322
323
324
        event_data = (
            f"event: {event_type}\ndata: {event.model_dump_json(indent=None)}\n\n"
        )
325
326
327
        yield event_data


328
329
330
331
332
333
334
335
336
337
@router.post(
    "/v1/responses",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
338
339
340
341
342
@with_cancellation
async def create_responses(request: ResponsesRequest, raw_request: Request):
    handler = responses(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
343
344
            message="The model does not support Responses API"
        )
345
346
347
    try:
        generator = await handler.create_responses(request, raw_request)
    except Exception as e:
348
349
350
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
351
352

    if isinstance(generator, ErrorResponse):
353
354
355
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
356
357
    elif isinstance(generator, ResponsesResponse):
        return JSONResponse(content=generator.model_dump())
358

359
360
361
    return StreamingResponse(
        content=_convert_stream_to_sse_events(generator), media_type="text/event-stream"
    )
362
363
364


@router.get("/v1/responses/{response_id}")
365
366
367
async def retrieve_responses(
    response_id: str,
    raw_request: Request,
368
369
    starting_after: int | None = None,
    stream: bool | None = False,
370
):
371
372
373
    handler = responses(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
374
375
            message="The model does not support Responses API"
        )
376

377
    try:
378
379
380
381
382
        response = await handler.retrieve_responses(
            response_id,
            starting_after=starting_after,
            stream=stream,
        )
383
    except Exception as e:
384
385
386
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
387
388

    if isinstance(response, ErrorResponse):
389
390
391
        return JSONResponse(
            content=response.model_dump(), status_code=response.error.code
        )
392
393
    elif isinstance(response, ResponsesResponse):
        return JSONResponse(content=response.model_dump())
394
395
396
    return StreamingResponse(
        content=_convert_stream_to_sse_events(response), media_type="text/event-stream"
    )
397
398
399
400
401
402
403


@router.post("/v1/responses/{response_id}/cancel")
async def cancel_responses(response_id: str, raw_request: Request):
    handler = responses(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
404
405
            message="The model does not support Responses API"
        )
406

407
408
409
    try:
        response = await handler.cancel_responses(response_id)
    except Exception as e:
410
411
412
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
413
414

    if isinstance(response, ErrorResponse):
415
416
417
        return JSONResponse(
            content=response.model_dump(), status_code=response.error.code
        )
418
419
420
    return JSONResponse(content=response.model_dump())


421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
@router.post(
    "/v1/messages",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": AnthropicErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": AnthropicErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": AnthropicErrorResponse},
    },
)
@with_cancellation
@load_aware_call
async def create_messages(request: AnthropicMessagesRequest, raw_request: Request):
    def translate_error_response(response: ErrorResponse) -> JSONResponse:
        anthropic_error = AnthropicErrorResponse(
            error=AnthropicError(
                type=response.error.type,
                message=response.error.message,
            )
        )
        return JSONResponse(
            status_code=response.error.code, content=anthropic_error.model_dump()
        )

    handler = messages(raw_request)
    if handler is None:
        error = base(raw_request).create_error_response(
            message="The model does not support Messages API"
        )
        return translate_error_response(error)

    try:
        generator = await handler.create_messages(request, raw_request)
    except Exception as e:
        logger.exception("Error in create_messages: %s", e)
        return JSONResponse(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
            content=AnthropicErrorResponse(
                error=AnthropicError(
                    type="internal_error",
                    message=str(e),
                )
            ).model_dump(),
        )

    if isinstance(generator, ErrorResponse):
        return translate_error_response(generator)

    elif isinstance(generator, AnthropicMessagesResponse):
470
471
472
        resp = generator.model_dump(exclude_none=True)
        logger.debug("Anthropic Messages Response: %s", resp)
        return JSONResponse(content=resp)
473
474
475
476

    return StreamingResponse(content=generator, media_type="text/event-stream")


477
478
479
480
481
482
483
484
485
486
@router.post(
    "/v1/chat/completions",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
487
@with_cancellation
488
@load_aware_call
489
async def create_chat_completion(request: ChatCompletionRequest, raw_request: Request):
490
491
492
    metrics_header_format = raw_request.headers.get(
        ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL, ""
    )
493
494
495
    handler = chat(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
496
497
            message="The model does not support Chat Completions API"
        )
498
499
500
    try:
        generator = await handler.create_chat_completion(request, raw_request)
    except Exception as e:
501
502
503
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
504
    if isinstance(generator, ErrorResponse):
505
506
507
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
508

509
    elif isinstance(generator, ChatCompletionResponse):
510
511
512
513
        return JSONResponse(
            content=generator.model_dump(),
            headers=metrics_header(metrics_header_format),
        )
514

515
516
    return StreamingResponse(content=generator, media_type="text/event-stream")

517

518
519
520
521
522
523
524
525
526
527
@router.post(
    "/v1/completions",
    dependencies=[Depends(validate_json_request)],
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
528
@with_cancellation
529
@load_aware_call
530
async def create_completion(request: CompletionRequest, raw_request: Request):
531
532
533
    metrics_header_format = raw_request.headers.get(
        ENDPOINT_LOAD_METRICS_FORMAT_HEADER_LABEL, ""
    )
534
535
536
    handler = completion(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
537
538
            message="The model does not support Completions API"
        )
539

540
541
542
    try:
        generator = await handler.create_completion(request, raw_request)
    except OverflowError as e:
543
544
545
        raise HTTPException(
            status_code=HTTPStatus.BAD_REQUEST.value, detail=str(e)
        ) from e
546
    except Exception as e:
547
548
549
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
550

551
    if isinstance(generator, ErrorResponse):
552
553
554
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
555
    elif isinstance(generator, CompletionResponse):
556
557
558
559
        return JSONResponse(
            content=generator.model_dump(),
            headers=metrics_header(metrics_header_format),
        )
Zhuohan Li's avatar
Zhuohan Li committed
560

561
562
    return StreamingResponse(content=generator, media_type="text/event-stream")

Zhuohan Li's avatar
Zhuohan Li committed
563

564
565
566
567
568
569
570
571
572
@router.post(
    "/v1/audio/transcriptions",
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.UNPROCESSABLE_ENTITY.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
573
@with_cancellation
574
@load_aware_call
575
576
577
async def create_transcriptions(
    raw_request: Request, request: Annotated[TranscriptionRequest, Form()]
):
578
579
580
    handler = transcription(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
581
582
            message="The model does not support Transcriptions API"
        )
583
584

    audio_data = await request.file.read()
585
    try:
586
        generator = await handler.create_transcription(audio_data, request, raw_request)
587
    except Exception as e:
588
589
590
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
591
592

    if isinstance(generator, ErrorResponse):
593
594
595
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
596

597
    elif isinstance(generator, TranscriptionResponseVariant):
598
599
600
601
602
        return JSONResponse(content=generator.model_dump())

    return StreamingResponse(content=generator, media_type="text/event-stream")


603
604
605
606
607
608
609
610
611
@router.post(
    "/v1/audio/translations",
    responses={
        HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
        HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
        HTTPStatus.UNPROCESSABLE_ENTITY.value: {"model": ErrorResponse},
        HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
    },
)
612
613
@with_cancellation
@load_aware_call
614
615
616
async def create_translations(
    request: Annotated[TranslationRequest, Form()], raw_request: Request
):
617
618
619
    handler = translation(raw_request)
    if handler is None:
        return base(raw_request).create_error_response(
620
621
            message="The model does not support Translations API"
        )
622
623

    audio_data = await request.file.read()
624
    try:
625
        generator = await handler.create_translation(audio_data, request, raw_request)
626
    except Exception as e:
627
628
629
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value, detail=str(e)
        ) from e
630
631

    if isinstance(generator, ErrorResponse):
632
633
634
        return JSONResponse(
            content=generator.model_dump(), status_code=generator.error.code
        )
635

636
    elif isinstance(generator, TranslationResponseVariant):
637
638
639
640
641
        return JSONResponse(content=generator.model_dump())

    return StreamingResponse(content=generator, media_type="text/event-stream")


642
if envs.VLLM_SERVER_DEV_MODE:
643
644
645
646
    logger.warning(
        "SECURITY WARNING: Development endpoints are enabled! "
        "This should NOT be used in production!"
    )
647

648
649
    PydanticVllmConfig = pydantic.TypeAdapter(VllmConfig)

650
    @router.get("/server_info")
651
652
    async def show_server_info(
        raw_request: Request,
653
        config_format: Annotated[Literal["text", "json"], Query()] = "text",
654
655
656
    ):
        vllm_config: VllmConfig = raw_request.app.state.vllm_config
        server_info = {
657
658
659
            "vllm_config": str(vllm_config)
            if config_format == "text"
            else PydanticVllmConfig.dump_python(vllm_config, mode="json", fallback=str)
660
661
            # fallback=str is needed to handle e.g. torch.dtype
        }
662
663
        return JSONResponse(content=server_info)

664
    @router.post("/reset_prefix_cache")
665
666
667
    async def reset_prefix_cache(
        raw_request: Request, reset_running_requests: bool = Query(default=False)
    ):
668
669
670
671
        """
        Reset the prefix cache. Note that we currently do not check if the
        prefix cache is successfully reset in the API server.
        """
672
        logger.info("Resetting prefix cache...")
673
        await engine_client(raw_request).reset_prefix_cache(reset_running_requests)
674
675
        return Response(status_code=200)

676
677
678
679
680
681
682
683
684
685
    @router.post("/reset_mm_cache")
    async def reset_mm_cache(raw_request: Request):
        """
        Reset the multi-modal cache. Note that we currently do not check if the
        multi-modal cache is successfully reset in the API server.
        """
        logger.info("Resetting multi-modal cache...")
        await engine_client(raw_request).reset_mm_cache()
        return Response(status_code=200)

686
687
688
689
690
    @router.post("/collective_rpc")
    async def collective_rpc(raw_request: Request):
        try:
            body = await raw_request.json()
        except json.JSONDecodeError as e:
691
692
693
694
            raise HTTPException(
                status_code=HTTPStatus.BAD_REQUEST.value,
                detail=f"JSON decode error: {e}",
            ) from e
695
696
        method = body.get("method")
        if method is None:
697
698
699
700
            raise HTTPException(
                status_code=HTTPStatus.BAD_REQUEST.value,
                detail="Missing 'method' in request body",
            )
701
        # For security reason, only serialized string args/kwargs are passed.
702
        # User-defined `method` is responsible for deserialization if needed.
703
704
        args: list[str] = body.get("args", [])
        kwargs: dict[str, str] = body.get("kwargs", {})
705
        timeout: float | None = body.get("timeout")
706
        results = await engine_client(raw_request).collective_rpc(
707
708
            method=method, timeout=timeout, args=tuple(args), kwargs=kwargs
        )
709
710
711
712
        if results is None:
            return Response(status_code=200)
        response: list[Any] = []
        for result in results:
713
            if result is None or isinstance(result, dict | list):
714
715
716
717
718
                response.append(result)
            else:
                response.append(str(result))
        return JSONResponse(content={"results": response})

719

720
def load_log_config(log_config_file: str | None) -> dict | None:
721
722
723
724
725
726
    if not log_config_file:
        return None
    try:
        with open(log_config_file) as f:
            return json.load(f)
    except Exception as e:
727
728
729
        logger.warning(
            "Failed to load log config from file %s: error %s", log_config_file, e
        )
730
731
732
        return None


733
734
735
class AuthenticationMiddleware:
    """
    Pure ASGI middleware that authenticates each request by checking
736
    if the Authorization Bearer token exists and equals anyof "{api_key}".
737
738
739
740
741
742
743
744

    Notes
    -----
    There are two cases in which authentication is skipped:
        1. The HTTP method is OPTIONS.
        2. The request path doesn't start with /v1 (e.g. /health).
    """

745
    def __init__(self, app: ASGIApp, tokens: list[str]) -> None:
746
        self.app = app
747
        self.api_tokens = [hashlib.sha256(t.encode("utf-8")).digest() for t in tokens]
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764

    def verify_token(self, headers: Headers) -> bool:
        authorization_header_value = headers.get("Authorization")
        if not authorization_header_value:
            return False

        scheme, _, param = authorization_header_value.partition(" ")
        if scheme.lower() != "bearer":
            return False

        param_hash = hashlib.sha256(param.encode("utf-8")).digest()

        token_match = False
        for token_hash in self.api_tokens:
            token_match |= secrets.compare_digest(param_hash, token_hash)

        return token_match
765

766
767
    def __call__(self, scope: Scope, receive: Receive, send: Send) -> Awaitable[None]:
        if scope["type"] not in ("http", "websocket") or scope["method"] == "OPTIONS":
768
769
770
771
772
773
774
            # scope["type"] can be "lifespan" or "startup" for example,
            # in which case we don't need to do anything
            return self.app(scope, receive, send)
        root_path = scope.get("root_path", "")
        url_path = URL(scope=scope).path.removeprefix(root_path)
        headers = Headers(scope=scope)
        # Type narrow to satisfy mypy.
775
        if url_path.startswith("/v1") and not self.verify_token(headers):
776
            response = JSONResponse(content={"error": "Unauthorized"}, status_code=401)
777
778
779
780
781
782
783
784
785
786
787
788
789
790
            return response(scope, receive, send)
        return self.app(scope, receive, send)


class XRequestIdMiddleware:
    """
    Middleware the set's the X-Request-Id header for each response
    to a random uuid4 (hex) value if the header isn't already
    present in the request, otherwise use the provided request id.
    """

    def __init__(self, app: ASGIApp) -> None:
        self.app = app

791
    def __call__(self, scope: Scope, receive: Receive, send: Send) -> Awaitable[None]:
792
793
794
795
796
797
798
799
800
801
802
803
804
        if scope["type"] not in ("http", "websocket"):
            return self.app(scope, receive, send)

        # Extract the request headers.
        request_headers = Headers(scope=scope)

        async def send_with_request_id(message: Message) -> None:
            """
            Custom send function to mutate the response headers
            and append X-Request-Id to it.
            """
            if message["type"] == "http.response.start":
                response_headers = MutableHeaders(raw=message["headers"])
805
                request_id = request_headers.get("X-Request-Id", uuid.uuid4().hex)
806
807
808
809
810
811
                response_headers.append("X-Request-Id", request_id)
            await send(message)

        return self.app(scope, receive, send_with_request_id)


812
813
814
815
def _extract_content_from_chunk(chunk_data: dict) -> str:
    """Extract content from a streaming response chunk."""
    try:
        from vllm.entrypoints.openai.protocol import (
816
817
818
            ChatCompletionStreamResponse,
            CompletionStreamResponse,
        )
819
820

        # Try using Completion types for type-safe parsing
821
822
        if chunk_data.get("object") == "chat.completion.chunk":
            chat_response = ChatCompletionStreamResponse.model_validate(chunk_data)
823
824
            if chat_response.choices and chat_response.choices[0].delta.content:
                return chat_response.choices[0].delta.content
825
826
827
        elif chunk_data.get("object") == "text_completion":
            completion_response = CompletionStreamResponse.model_validate(chunk_data)
            if completion_response.choices and completion_response.choices[0].text:
828
829
830
                return completion_response.choices[0].text
    except pydantic.ValidationError:
        # Fallback to manual parsing
831
832
833
834
835
836
        if "choices" in chunk_data and chunk_data["choices"]:
            choice = chunk_data["choices"][0]
            if "delta" in choice and choice["delta"].get("content"):
                return choice["delta"]["content"]
            elif choice.get("text"):
                return choice["text"]
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
    return ""


class SSEDecoder:
    """Robust Server-Sent Events decoder for streaming responses."""

    def __init__(self):
        self.buffer = ""
        self.content_buffer = []

    def decode_chunk(self, chunk: bytes) -> list[dict]:
        """Decode a chunk of SSE data and return parsed events."""
        import json

        try:
852
            chunk_str = chunk.decode("utf-8")
853
854
855
856
857
858
859
860
        except UnicodeDecodeError:
            # Skip malformed chunks
            return []

        self.buffer += chunk_str
        events = []

        # Process complete lines
861
862
863
        while "\n" in self.buffer:
            line, self.buffer = self.buffer.split("\n", 1)
            line = line.rstrip("\r")  # Handle CRLF
864

865
            if line.startswith("data: "):
866
                data_str = line[6:].strip()
867
868
                if data_str == "[DONE]":
                    events.append({"type": "done"})
869
870
871
                elif data_str:
                    try:
                        event_data = json.loads(data_str)
872
                        events.append({"type": "data", "data": event_data})
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
                    except json.JSONDecodeError:
                        # Skip malformed JSON
                        continue

        return events

    def extract_content(self, event_data: dict) -> str:
        """Extract content from event data."""
        return _extract_content_from_chunk(event_data)

    def add_content(self, content: str) -> None:
        """Add content to the buffer."""
        if content:
            self.content_buffer.append(content)

    def get_complete_content(self) -> str:
        """Get the complete buffered content."""
890
        return "".join(self.content_buffer)
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910


def _log_streaming_response(response, response_body: list) -> None:
    """Log streaming response with robust SSE parsing."""
    from starlette.concurrency import iterate_in_threadpool

    sse_decoder = SSEDecoder()
    chunk_count = 0

    def buffered_iterator():
        nonlocal chunk_count

        for chunk in response_body:
            chunk_count += 1
            yield chunk

            # Parse SSE events from chunk
            events = sse_decoder.decode_chunk(chunk)

            for event in events:
911
912
                if event["type"] == "data":
                    content = sse_decoder.extract_content(event["data"])
913
                    sse_decoder.add_content(content)
914
                elif event["type"] == "done":
915
916
917
918
919
920
921
922
                    # Log complete content when done
                    full_content = sse_decoder.get_complete_content()
                    if full_content:
                        # Truncate if too long
                        if len(full_content) > 2048:
                            full_content = full_content[:2048] + ""
                            "...[truncated]"
                        logger.info(
923
                            "response_body={streaming_complete: content=%r, chunks=%d}",
924
925
926
                            full_content,
                            chunk_count,
                        )
927
928
                    else:
                        logger.info(
929
930
931
                            "response_body={streaming_complete: no_content, chunks=%d}",
                            chunk_count,
                        )
932
933
934
                    return

    response.body_iterator = iterate_in_threadpool(buffered_iterator())
935
    logger.info("response_body={streaming_started: chunks=%d}", len(response_body))
936
937
938
939
940
941
942
943
944
945
946


def _log_non_streaming_response(response_body: list) -> None:
    """Log non-streaming response."""
    try:
        decoded_body = response_body[0].decode()
        logger.info("response_body={%s}", decoded_body)
    except UnicodeDecodeError:
        logger.info("response_body={<binary_data>}")


947
def build_app(args: Namespace) -> FastAPI:
948
    if args.disable_fastapi_docs:
949
950
951
        app = FastAPI(
            openapi_url=None, docs_url=None, redoc_url=None, lifespan=lifespan
        )
952
953
    else:
        app = FastAPI(lifespan=lifespan)
954
955
    app.state.args = args
    from vllm.entrypoints.serve import register_vllm_serve_api_routers
956

957
    register_vllm_serve_api_routers(app)
958
959
960
961

    from vllm.entrypoints.sagemaker.routes import register_sagemaker_routes

    register_sagemaker_routes(router)
Ethan Xu's avatar
Ethan Xu committed
962
    app.include_router(router)
963

Ethan Xu's avatar
Ethan Xu committed
964
    app.root_path = args.root_path
Zhuohan Li's avatar
Zhuohan Li committed
965

966
967
968
969
    from vllm.entrypoints.pooling import register_pooling_api_routers

    register_pooling_api_routers(app)

Zhuohan Li's avatar
Zhuohan Li committed
970
971
972
973
974
975
976
977
    app.add_middleware(
        CORSMiddleware,
        allow_origins=args.allowed_origins,
        allow_credentials=args.allow_credentials,
        allow_methods=args.allowed_methods,
        allow_headers=args.allowed_headers,
    )

978
979
    @app.exception_handler(HTTPException)
    async def http_exception_handler(_: Request, exc: HTTPException):
980
        err = ErrorResponse(
981
982
983
984
985
986
            error=ErrorInfo(
                message=exc.detail,
                type=HTTPStatus(exc.status_code).phrase,
                code=exc.status_code,
            )
        )
987
988
        return JSONResponse(err.model_dump(), status_code=exc.status_code)

Ethan Xu's avatar
Ethan Xu committed
989
    @app.exception_handler(RequestValidationError)
990
    async def validation_exception_handler(_: Request, exc: RequestValidationError):
991
992
993
994
995
996
997
998
        exc_str = str(exc)
        errors_str = str(exc.errors())

        if exc.errors() and errors_str and errors_str != exc_str:
            message = f"{exc_str} {errors_str}"
        else:
            message = exc_str

999
1000
1001
1002
1003
1004
1005
1006
        err = ErrorResponse(
            error=ErrorInfo(
                message=message,
                type=HTTPStatus.BAD_REQUEST.phrase,
                code=HTTPStatus.BAD_REQUEST,
            )
        )
        return JSONResponse(err.model_dump(), status_code=HTTPStatus.BAD_REQUEST)
Ethan Xu's avatar
Ethan Xu committed
1007

1008
    # Ensure --api-key option from CLI takes precedence over VLLM_API_KEY
1009
1010
    if tokens := [key for key in (args.api_key or [envs.VLLM_API_KEY]) if key]:
        app.add_middleware(AuthenticationMiddleware, tokens=tokens)
1011

1012
    if args.enable_request_id_headers:
1013
        app.add_middleware(XRequestIdMiddleware)
1014

1015
1016
1017
    # Add scaling middleware to check for scaling state
    app.add_middleware(ScalingMiddleware)

1018
    if envs.VLLM_DEBUG_LOG_API_SERVER_RESPONSE:
1019
1020
1021
1022
1023
        logger.warning(
            "CAUTION: Enabling log response in the API Server. "
            "This can include sensitive information and should be "
            "avoided in production."
        )
1024
1025
1026
1027

        @app.middleware("http")
        async def log_response(request: Request, call_next):
            response = await call_next(request)
1028
            response_body = [section async for section in response.body_iterator]
1029
            response.body_iterator = iterate_in_threadpool(iter(response_body))
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
            # Check if this is a streaming response by looking at content-type
            content_type = response.headers.get("content-type", "")
            is_streaming = content_type == "text/event-stream; charset=utf-8"

            # Log response body based on type
            if not response_body:
                logger.info("response_body={<empty>}")
            elif is_streaming:
                _log_streaming_response(response, response_body)
            else:
                _log_non_streaming_response(response_body)
1041
            return response
1042

1043
1044
1045
1046
    for middleware in args.middleware:
        module_path, object_name = middleware.rsplit(".", 1)
        imported = getattr(importlib.import_module(module_path), object_name)
        if inspect.isclass(imported):
1047
            app.add_middleware(imported)  # type: ignore[arg-type]
1048
1049
1050
        elif inspect.iscoroutinefunction(imported):
            app.middleware("http")(imported)
        else:
1051
1052
1053
            raise ValueError(
                f"Invalid middleware {middleware}. Must be a function or a class."
            )
1054

1055
1056
    app = sagemaker_standards.bootstrap(app)

Ethan Xu's avatar
Ethan Xu committed
1057
1058
1059
    return app


1060
async def init_app_state(
1061
    engine_client: EngineClient,
1062
    state: State,
1063
    args: Namespace,
1064
) -> None:
1065
1066
    vllm_config = engine_client.vllm_config

1067
    if args.served_model_name is not None:
1068
        served_model_names = args.served_model_name
1069
    else:
1070
        served_model_names = [args.model]
1071

1072
    if args.enable_log_requests:
1073
        request_logger = RequestLogger(max_log_len=args.max_log_len)
1074
1075
    else:
        request_logger = None
1076

1077
    base_model_paths = [
1078
        BaseModelPath(name=name, model_path=args.model) for name in served_model_names
1079
1080
    ]

1081
    state.engine_client = engine_client
1082
    state.log_stats = not args.disable_log_stats
1083
    state.vllm_config = vllm_config
1084
    state.args = args
1085
    supported_tasks = await engine_client.get_supported_tasks()
1086
    logger.info("Supported tasks: %s", supported_tasks)
1087

1088
1089
1090
    resolved_chat_template = await process_chat_template(
        args.chat_template, engine_client, vllm_config.model_config
    )
1091

1092
    if args.tool_server == "demo":
1093
        tool_server: ToolServer | None = DemoToolServer()
1094
1095
        assert isinstance(tool_server, DemoToolServer)
        await tool_server.init_and_validate()
1096
1097
1098
    elif args.tool_server:
        tool_server = MCPToolServer()
        await tool_server.add_tool_server(args.tool_server)
1099
1100
1101
    else:
        tool_server = None

1102
    # Merge default_mm_loras into the static lora_modules
1103
1104
1105
1106
1107
    default_mm_loras = (
        vllm_config.lora_config.default_mm_loras
        if vllm_config.lora_config is not None
        else {}
    )
1108

1109
1110
1111
1112
1113
1114
    default_mm_loras = (
        vllm_config.lora_config.default_mm_loras
        if vllm_config.lora_config is not None
        else {}
    )
    lora_modules = process_lora_modules(args.lora_modules, default_mm_loras)
1115

1116
    state.openai_serving_models = OpenAIServingModels(
1117
        engine_client=engine_client,
1118
        base_model_paths=base_model_paths,
1119
        lora_modules=lora_modules,
1120
    )
1121
    await state.openai_serving_models.init_static_loras()
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
    state.openai_serving_responses = (
        OpenAIServingResponses(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            enable_auto_tools=args.enable_auto_tool_choice,
            tool_parser=args.tool_call_parser,
            tool_server=tool_server,
            reasoning_parser=args.structured_outputs_config.reasoning_parser,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_force_include_usage=args.enable_force_include_usage,
            enable_log_outputs=args.enable_log_outputs,
            log_error_stack=args.log_error_stack,
        )
        if "generate" in supported_tasks
        else None
    )
    state.openai_serving_chat = (
        OpenAIServingChat(
            engine_client,
            state.openai_serving_models,
            args.response_role,
            request_logger=request_logger,
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            trust_request_chat_template=args.trust_request_chat_template,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            enable_auto_tools=args.enable_auto_tool_choice,
            exclude_tools_when_tool_choice_none=args.exclude_tools_when_tool_choice_none,
            tool_parser=args.tool_call_parser,
            reasoning_parser=args.structured_outputs_config.reasoning_parser,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_force_include_usage=args.enable_force_include_usage,
            enable_log_outputs=args.enable_log_outputs,
            log_error_stack=args.log_error_stack,
        )
        if "generate" in supported_tasks
        else None
    )
    state.openai_serving_completion = (
        OpenAIServingCompletion(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_force_include_usage=args.enable_force_include_usage,
            log_error_stack=args.log_error_stack,
        )
        if "generate" in supported_tasks
        else None
    )
    state.openai_serving_pooling = (
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
        (
            OpenAIServingPooling(
                engine_client,
                state.openai_serving_models,
                supported_tasks=supported_tasks,
                request_logger=request_logger,
                chat_template=resolved_chat_template,
                chat_template_content_format=args.chat_template_content_format,
                trust_request_chat_template=args.trust_request_chat_template,
                log_error_stack=args.log_error_stack,
            )
1189
        )
1190
        if any(task in POOLING_TASKS for task in supported_tasks)
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
        else None
    )
    state.openai_serving_embedding = (
        OpenAIServingEmbedding(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            trust_request_chat_template=args.trust_request_chat_template,
            log_error_stack=args.log_error_stack,
        )
        if "embed" in supported_tasks
        else None
    )
    state.openai_serving_classification = (
        ServingClassification(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
1211
1212
1213
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            trust_request_chat_template=args.trust_request_chat_template,
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
            log_error_stack=args.log_error_stack,
        )
        if "classify" in supported_tasks
        else None
    )
    state.openai_serving_scores = (
        ServingScores(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            log_error_stack=args.log_error_stack,
        )
        if ("embed" in supported_tasks or "score" in supported_tasks)
        else None
    )
1229
    state.openai_serving_tokenization = OpenAIServingTokenization(
1230
        engine_client,
1231
        state.openai_serving_models,
1232
        request_logger=request_logger,
1233
1234
        chat_template=resolved_chat_template,
        chat_template_content_format=args.chat_template_content_format,
1235
        trust_request_chat_template=args.trust_request_chat_template,
1236
        log_error_stack=args.log_error_stack,
1237
    )
1238
1239
1240
1241
1242
1243
    state.openai_serving_transcription = (
        OpenAIServingTranscription(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            log_error_stack=args.log_error_stack,
1244
            enable_force_include_usage=args.enable_force_include_usage,
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
        )
        if "transcription" in supported_tasks
        else None
    )
    state.openai_serving_translation = (
        OpenAIServingTranslation(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            log_error_stack=args.log_error_stack,
1255
            enable_force_include_usage=args.enable_force_include_usage,
1256
1257
1258
1259
        )
        if "transcription" in supported_tasks
        else None
    )
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
    state.anthropic_serving_messages = (
        AnthropicServingMessages(
            engine_client,
            state.openai_serving_models,
            args.response_role,
            request_logger=request_logger,
            chat_template=resolved_chat_template,
            chat_template_content_format=args.chat_template_content_format,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            enable_auto_tools=args.enable_auto_tool_choice,
            tool_parser=args.tool_call_parser,
            reasoning_parser=args.structured_outputs_config.reasoning_parser,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_force_include_usage=args.enable_force_include_usage,
        )
        if "generate" in supported_tasks
        else None
    )
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
    state.serving_tokens = (
        ServingTokens(
            engine_client,
            state.openai_serving_models,
            request_logger=request_logger,
            return_tokens_as_token_ids=args.return_tokens_as_token_ids,
            log_error_stack=args.log_error_stack,
            enable_prompt_tokens_details=args.enable_prompt_tokens_details,
            enable_log_outputs=args.enable_log_outputs,
            force_no_detokenize=args.tokens_only,
        )
        if "generate" in supported_tasks
        else None
    )
1292

1293
1294
1295
    state.enable_server_load_tracking = args.enable_server_load_tracking
    state.server_load_metrics = 0

1296

1297
def create_server_socket(addr: tuple[str, int]) -> socket.socket:
1298
1299
1300
1301
1302
1303
    family = socket.AF_INET
    if is_valid_ipv6_address(addr[0]):
        family = socket.AF_INET6

    sock = socket.socket(family=family, type=socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1304
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
1305
1306
1307
1308
1309
    sock.bind(addr)

    return sock


1310
1311
1312
1313
1314
1315
def create_server_unix_socket(path: str) -> socket.socket:
    sock = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM)
    sock.bind(path)
    return sock


1316
def validate_api_server_args(args):
1317
    valid_tool_parses = ToolParserManager.list_registered()
1318
1319
1320
1321
1322
    if args.enable_auto_tool_choice and args.tool_call_parser not in valid_tool_parses:
        raise KeyError(
            f"invalid tool call parser: {args.tool_call_parser} "
            f"(chose from {{ {','.join(valid_tool_parses)} }})"
        )
1323

1324
    valid_reasoning_parsers = ReasoningParserManager.list_registered()
1325
1326
    if (
        reasoning_parser := args.structured_outputs_config.reasoning_parser
1327
    ) and reasoning_parser not in valid_reasoning_parsers:
1328
        raise KeyError(
1329
            f"invalid reasoning parser: {reasoning_parser} "
1330
            f"(chose from {{ {','.join(valid_reasoning_parsers)} }})"
1331
        )
1332

1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343

def setup_server(args):
    """Validate API server args, set up signal handler, create socket
    ready to serve."""

    logger.info("vLLM API server version %s", VLLM_VERSION)
    log_non_default_args(args)

    if args.tool_parser_plugin and len(args.tool_parser_plugin) > 3:
        ToolParserManager.import_tool_parser(args.tool_parser_plugin)

1344
1345
1346
    if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3:
        ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin)

1347
1348
    validate_api_server_args(args)

1349
1350
1351
    # workaround to make sure that we bind the port before the engine is set up.
    # This avoids race conditions with ray.
    # see https://github.com/vllm-project/vllm/issues/8204
1352
1353
1354
1355
1356
    if args.uds:
        sock = create_server_unix_socket(args.uds)
    else:
        sock_addr = (args.host or "", args.port)
        sock = create_server_socket(sock_addr)
1357

1358
1359
1360
1361
    # workaround to avoid footguns where uvicorn drops requests with too
    # many concurrent requests active
    set_ulimit()

1362
1363
1364
1365
1366
1367
    def signal_handler(*_) -> None:
        # Interrupt server on sigterm while initializing
        raise KeyboardInterrupt("terminated")

    signal.signal(signal.SIGTERM, signal_handler)

1368
1369
1370
1371
1372
    if args.uds:
        listen_address = f"unix:{args.uds}"
    else:
        addr, port = sock_addr
        is_ssl = args.ssl_keyfile and args.ssl_certfile
1373
        host_part = f"[{addr}]" if is_valid_ipv6_address(addr) else addr or "0.0.0.0"
1374
        listen_address = f"http{'s' if is_ssl else ''}://{host_part}:{port}"
1375
1376
1377
1378
1379
    return listen_address, sock


async def run_server(args, **uvicorn_kwargs) -> None:
    """Run a single-worker API server."""
1380
1381

    # Add process-specific prefix to stdout and stderr.
1382
    decorate_logs("APIServer")
1383

1384
1385
1386
1387
    listen_address, sock = setup_server(args)
    await run_server_worker(listen_address, sock, args, **uvicorn_kwargs)


1388
1389
1390
async def run_server_worker(
    listen_address, sock, args, client_config=None, **uvicorn_kwargs
) -> None:
1391
1392
1393
1394
1395
    """Run a single API server worker."""

    if args.tool_parser_plugin and len(args.tool_parser_plugin) > 3:
        ToolParserManager.import_tool_parser(args.tool_parser_plugin)

1396
1397
1398
    if args.reasoning_parser_plugin and len(args.reasoning_parser_plugin) > 3:
        ReasoningParserManager.import_reasoning_parser(args.reasoning_parser_plugin)

1399
1400
1401
    # Load logging config for uvicorn if specified
    log_config = load_log_config(args.log_config_file)
    if log_config is not None:
1402
        uvicorn_kwargs["log_config"] = log_config
1403

1404
    async with build_async_engine_client(
1405
1406
        args,
        client_config=client_config,
1407
    ) as engine_client:
1408
1409
        app = build_app(args)

1410
        await init_app_state(engine_client, app.state, args)
1411

1412
1413
        logger.info(
            "Starting vLLM API server %d on %s",
1414
            engine_client.vllm_config.parallel_config._api_process_rank,
1415
1416
            listen_address,
        )
1417
1418
        shutdown_task = await serve_http(
            app,
1419
            sock=sock,
1420
            enable_ssl_refresh=args.enable_ssl_refresh,
1421
1422
1423
            host=args.host,
            port=args.port,
            log_level=args.uvicorn_log_level,
1424
1425
1426
            # NOTE: When the 'disable_uvicorn_access_log' value is True,
            # no access log will be output.
            access_log=not args.disable_uvicorn_access_log,
1427
            timeout_keep_alive=envs.VLLM_HTTP_TIMEOUT_KEEP_ALIVE,
1428
1429
1430
1431
            ssl_keyfile=args.ssl_keyfile,
            ssl_certfile=args.ssl_certfile,
            ssl_ca_certs=args.ssl_ca_certs,
            ssl_cert_reqs=args.ssl_cert_reqs,
1432
1433
            h11_max_incomplete_event_size=args.h11_max_incomplete_event_size,
            h11_max_header_count=args.h11_max_header_count,
1434
1435
1436
            **uvicorn_kwargs,
        )

1437
    # NB: Await server shutdown only after the backend context is exited
1438
1439
1440
1441
    try:
        await shutdown_task
    finally:
        sock.close()
1442

Ethan Xu's avatar
Ethan Xu committed
1443
1444
1445

if __name__ == "__main__":
    # NOTE(simon):
1446
1447
    # This section should be in sync with vllm/entrypoints/cli/main.py for CLI
    # entrypoints.
1448
    cli_env_setup()
Ethan Xu's avatar
Ethan Xu committed
1449
    parser = FlexibleArgumentParser(
1450
1451
        description="vLLM OpenAI-Compatible RESTful API server."
    )
Ethan Xu's avatar
Ethan Xu committed
1452
1453
    parser = make_arg_parser(parser)
    args = parser.parse_args()
1454
    validate_parsed_serve_args(args)
1455

1456
    uvloop.run(run_server(args))