def METHOD_NAME(
    port: Optional[int] = None,
    socket: Optional[str] = None,
    host: str = "localhost",
    max_workers: Optional[int] = None,
    fixed_server_id: Optional[str] = None,
    log_level: str = "INFO",
    use_python_environment_entry_point: bool = False,
    container_image: Optional[str] = None,
    container_context: Optional[str] = None,
    location_name: Optional[str] = None,
    inject_env_vars_from_instance: bool = False,
    startup_timeout: int = 0,
    instance_ref=None,
    **kwargs,
):
    from dagster._grpc import DagsterGrpcServer
    from dagster._grpc.proxy_server import DagsterProxyApiServicer
    if seven.IS_WINDOWS and port is None:
        raise click.UsageError(
            "You must pass a valid --port/-p on Windows: --socket/-s not supported."
        )
    if not (port or socket and not (port and socket)):
        raise click.UsageError("You must pass one and only one of --port/-p or --socket/-s.")
    setup_interrupt_handlers()
    configure_loggers(log_level=log_level.upper())
    logger = logging.getLogger("dagster.code_server")
    container_image = container_image or os.getenv("DAGSTER_CURRENT_IMAGE")
    # in the gRPC api CLI we never load more than one module or python file at a time
    module_name = check.opt_str_elem(kwargs, "module_name")
    python_file = check.opt_str_elem(kwargs, "python_file")
    loadable_target_origin = LoadableTargetOrigin(
        executable_path=sys.executable if use_python_environment_entry_point else None,
        attribute=kwargs["attribute"],
        working_directory=get_working_directory_from_kwargs(kwargs),
        module_name=module_name,
        python_file=python_file,
        package_name=kwargs["package_name"],
    )
    server_termination_event = threading.Event()
    api_servicer = DagsterProxyApiServicer(
        loadable_target_origin=loadable_target_origin,
        fixed_server_id=fixed_server_id,
        container_image=container_image,
        container_context=(
            json.loads(container_context) if container_context is not None else None
        ),
        inject_env_vars_from_instance=inject_env_vars_from_instance,
        location_name=location_name,
        log_level=log_level,
        startup_timeout=startup_timeout,
        instance_ref=deserialize_value(instance_ref, InstanceRef) if instance_ref else None,
        server_termination_event=server_termination_event,
        logger=logger,
    )
    server = DagsterGrpcServer(
        server_termination_event=server_termination_event,
        dagster_api_servicer=api_servicer,
        port=port,
        socket=socket,
        host=host,
        max_workers=max_workers,
        logger=logger,
    )
    code_desc = " "
    if loadable_target_origin.python_file:
        code_desc = f" for file {loadable_target_origin.python_file} "
    elif loadable_target_origin.package_name:
        code_desc = f" for package {loadable_target_origin.package_name} "
    elif loadable_target_origin.module_name:
        code_desc = f" for module {loadable_target_origin.module_name} "
    server_desc = (
        f"Dagster code proxy server{code_desc}on port {port} in process {os.getpid()}"
        if port
        else f"Dagster code proxy server{code_desc}in process {os.getpid()}"
    )
    logger.info("Started %s", server_desc)
    try:
        server.serve()
    except KeyboardInterrupt:
        # Terminate cleanly on interrupt
        logger.info("Code proxy server was interrupted")
    finally:
        logger.info("Shutting down %s", server_desc)