Function vc_handler

python/vercel-runtime/src/vercel_runtime/vc_init.py:1099–1149

(event: dict[str, Any], context: Any)

    port = server.server_address[1]  # type: ignore[attr-defined]

    def vc_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
        payload = json.loads(event["body"])
        path, _ = apply_service_route_prefix_to_target(payload["path"])
        headers = normalize_event_headers(payload.get("headers", {}))
        method = payload["method"]
        encoding = payload.get("encoding")
        body = payload.get("body")

        sc_no_header_leak = bool(headers.get(SC_NO_HEADER_LEAK_HEADER))
        set_runtime_cache_from_http_headers(headers)
        for sc_header in SC_HEADERS_ALWAYS_STRIP:
            headers.pop(sc_header, None)
        if sc_no_header_leak:
            for sc_header in SC_HEADERS_STRIP_ON_NO_LEAK:
                headers.pop(sc_header, None)

        # `_thread.start_new_thread` does not propagate contextvars
        captured_ctx = contextvars.copy_context()
        _thread.start_new_thread(
            captured_ctx.run,
            (server.handle_request,),  # type: ignore[attr-defined]
        )
        clear_runtime_cache_context()

        if (body is not None and len(body) > 0) and (
            encoding is not None and encoding == "base64"
        ):
            body = base64.b64decode(body)

        request_body = body.encode("utf-8") if isinstance(body, str) else body
        conn = http.client.HTTPConnection("127.0.0.1", port)
        try:
            conn.request(method, path, headers=headers, body=request_body)
        except (OSError, http.client.HTTPException) as ex:
            _stderr(f"Request Error: {ex}")
        res = conn.getresponse()

        return_dict: dict[str, Any] = {
            "statusCode": res.status,
            "headers": format_headers(res.headers),
        }

        data = res.read()

        try:
            return_dict["body"] = data.decode("utf-8")
        except UnicodeDecodeError:
            return_dict["body"] = base64.b64encode(data).decode("utf-8")
            return_dict["encoding"] = "base64"

        return return_dict

else:
    try: