Spaces:
Sleeping
Sleeping
File size: 3,229 Bytes
2415446 a1bab2d 2415446 a1bab2d 2415446 a1bab2d 2415446 61bb677 a1bab2d 2415446 a1bab2d 2415446 a1bab2d 2415446 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | """Single production composition root for the FCC server."""
import os
from functools import partial
from pathlib import Path
from free_claude_code.api.app import create_app
from free_claude_code.api.ports import ApiServices
from free_claude_code.config.logging_config import configure_logging
from free_claude_code.config.paths import server_log_path
from free_claude_code.config.settings import Settings
from free_claude_code.messaging.transcription import TranscriptionService
from free_claude_code.messaging.voice import Transcriber
from free_claude_code.providers.admission import ProviderAdmissionController
from free_claude_code.providers.base import BaseProvider, ProviderConfig
from free_claude_code.providers.nvidia_nim.voice import NvidiaNimTranscriber
from free_claude_code.providers.openai_codex import (
OpenAIAuthManager,
OpenAICodexProvider,
)
from free_claude_code.providers.runtime import ProviderRuntime
from free_claude_code.providers.runtime.factory import create_provider
from .application import ApplicationRuntime, RestartCallback
from .asgi import RuntimeASGIApp
from .provider_manager import ProviderRuntimeManager
def build_asgi_app(
settings: Settings,
restart_callback: RestartCallback | None = None,
) -> RuntimeASGIApp:
"""Construct the complete server application and its resource owner."""
log_path = Path(os.getenv("LOG_FILE", server_log_path()))
configure_logging(
log_path,
level=settings.log_level,
verbose_third_party=settings.log_raw_api_payloads,
)
openai_auth = OpenAIAuthManager(proxy=settings.openai_proxy)
openai_factory = partial(_create_openai_provider, auth=openai_auth)
provider_constructor = partial(
create_provider,
injected_factories={"openai": openai_factory},
)
runtime_factory = partial(
ProviderRuntime,
provider_constructor=provider_constructor,
)
provider_manager = ProviderRuntimeManager(
settings,
runtime_factory=runtime_factory,
connected_provider_ids=openai_auth.connected_provider_ids,
)
runtime = ApplicationRuntime(
provider_manager,
transcriber=_create_transcriber(settings),
restart_callback=restart_callback,
connected_accounts={"openai": openai_auth},
)
services = ApiServices(
requests=provider_manager,
admin=runtime,
tasks=runtime,
)
return RuntimeASGIApp(create_app(services), runtime)
def _create_openai_provider(
config: ProviderConfig,
_settings: Settings,
admission: ProviderAdmissionController,
*,
auth: OpenAIAuthManager,
) -> BaseProvider:
return OpenAICodexProvider(config, auth=auth, admission=admission)
def _create_transcriber(settings: Settings) -> Transcriber | None:
if not settings.voice_note_enabled:
return None
if settings.whisper_device == "nvidia_nim":
return NvidiaNimTranscriber(
model=settings.whisper_model,
api_key=settings.nvidia_nim_api_key,
)
return TranscriptionService(
model=settings.whisper_model,
device=settings.whisper_device,
huggingface_api_key=settings.huggingface_api_key,
)
|