OpenAI Scala Client

Adapters allow you to wrap, distribute, or transform OpenAI services. They can be arbitrarily combined or stacked to provide features like load balancing, logging, or retries.

Common adapter types include:

• Load Distribution: roundRobin or randomOrder to spread requests across multiple service instances.
• Observability: log to intercept and log function calls.
• Resilience: retry to handle transient errors (like rate limits) using RetrySettings.
• Routing: chatCompletionRouter to route requests to different providers based on the requested model ID.

// Example: Round Robin Load Distribution
val adapters = OpenAIServiceAdapters.forFullService
val service1 = OpenAIServiceFactory("your-api-key1")
val service2 = OpenAIServiceFactory("your-api-key2")
val service = adapters.roundRobin(service1, service2)

// Example: Logging
val rawService = OpenAIServiceFactory()
val service = adapters.log(rawService, "openAIService", logger.log)

// Example: Retry on transient errors
implicit val retrySettings: RetrySettings = RetrySettings(maxRetries = 10).constantInterval(10.seconds)
val service = adapters.retry(OpenAIServiceFactory(), Some(println(_)))

Groq can be used via OpenAIChatCompletionServiceFactory(ChatProviderSettings.groq).

Important Implementation Details for Groq:

• Schema Validation: The schema is validated server-side. If a generation violates the schema, the server returns an HTTP 400 json_validate_failed error containing the failed_generation text. This should be treated as a retryable outcome.
• Tool/JSON Mode Conflict: You cannot combine response_format (JSON mode) with a non-empty tools array. To succeed while using tools, set tool_choice = "none".
• Model Support: Strict json_schema and function tools work on openai/gpt-oss-* and qwen/qwen3.x-27b models. Agentic models like groq/compound* and allam-2-7b reject these parameters.

// Standard Groq service
val service = OpenAIChatCompletionServiceFactory(ChatProviderSettings.groq)

// Streaming Groq service
val service = OpenAIChatCompletionServiceFactory.withStreaming(ChatProviderSettings.groq)

By default, every factory call (e.g., OpenAIServiceFactory()) creates its own HTTP client pool and dedicated actor system. To avoid the overhead of multiple pools and actor systems, you can create a single engine using StreamedEngineRegistry.outputStreamed() and share it across multiple providers (OpenAI, Anthropic, Gemini, etc.).

Lifecycle Management:

• service.close(): Closes the specific service's connection but does not touch the shared engine.
• engine.close(): The final teardown step. Call this only after all services using the engine are finished.

import io.cequence.wsclient.service.spi.StreamedEngineRegistry

implicit val ec: ExecutionContext = ExecutionContext.global

val engine = StreamedEngineRegistry.outputStreamed() // one pool + one (daemon) actor system

val openAI = OpenAIServiceFactory.withEngine(engine)
val anthropic = AnthropicServiceFactory.withEngine(engine)
val gemini = GeminiServiceFactory.withEngine(engine)

// ... use the services ...

anthropic.close() // closes a service on a SHARED engine without touching the engine itself
engine.close()     // the one real teardown

The client provides a uniform way to attach FileContent (e.g., PDFs) and ImageURLContent (e.g., JPEG, PNG, WebP) across OpenAI, Anthropic, Gemini, and Vertex AI.

To ensure models can refer to files by name, use VLMContent.of(bytes, name). This wraps the content in an envelope that includes a [file: NAME] label, allowing the model to use the exact filename in its response.

import io.cequence.openaiscala.domain.VLMContent

val files: Seq[(String, Array[Byte])] = ... // (fileName, bytes) - PDFs and images

val messages = Seq(
    SystemMessage(
      "Each attached file is preceded by a label of the form '[file: NAME]'. " +
        "When referring to a file in your answer, use exactly the NAME from its label."
    ),
    UserSeqMessage(TextContent("Summarize each attached file in one sentence.") +:
        files.flatMap { case (name, bytes) => VLMContent.of(bytes, name) }
    )
  )

service.createChatCompletion(
    messages = messages,
    settings = CreateChatCompletionSettings(model = NonOpenAIModelId.claude_sonnet_5)
  )

Instead of using provider-specific settings, you can pass MCPServerTool or SkillTool directly into the tools sequence of createChatToolCompletion or createChatToolCompletionStreamed. The client adapter will automatically map these to the appropriate native feature for the target provider (OpenAI, Anthropic, or Gemini).

import io.cequence.openaiscala.domain.ChatCompletionTool.{MCPServerTool, SkillTool, SkillSource}

service.createChatToolCompletionStreamed(
    messages = Seq(UserMessage("What is the 'given' keyword for in Scala 3? Check scala/scala3.")),
    tools = Seq(
      MCPServerTool("deepwiki", "https://mcp.deepwiki.com/mcp"),          // the provider calls it
      SkillTool("pptx", version = Some("latest"), source = SkillSource.Provider) // Anthropic's built-in skill
    ),
    settings = CreateChatCompletionSettings(NonOpenAIModelId.claude_sonnet_5)
  )

Amazon Bedrock can be accessed via OpenAIServiceFactory.forBedrock. It supports both the OpenAI Responses API and Chat Completions.

Key Configuration Options:

• isOpenAIModel: If true, models are served from the openai/v1 base path (e.g., openai.gpt-5.5). If false, they use the standard v1 path.
• auth: Determines authentication. Defaults to BedrockAuth.fromEnv() (using AWS_BEARER_TOKEN_BEDROCK). You can also use BedrockAuth.SigV4 with static credentials.
• endpoint: Specifies the host, such as BedrockEndpoint.Runtime.
• AWS_BEDROCK_REGION: Must be set in the environment.

Note: When using SigV4, rotating credentials are re-read automatically.

// Using OpenAI provider models via Bedrock
val service = OpenAIServiceFactory.forBedrock(isOpenAIModel = true)
service.createModelResponse(Inputs.Text("What is the capital of France?"),
  settings = CreateModelResponseSettings(model = ModelId.bedrock_openai_gpt_5_5))

// Using standard models via Bedrock
val service = OpenAIServiceFactory.forBedrock()

// Using explicit IAM credentials (SigV4)
val signed = OpenAIServiceFactory.forBedrock(
  auth = BedrockAuth.SigV4(AwsCredentialsProvider.static(accessKey, secretKey)),
  isOpenAIModel = true)

// Using Bedrock Runtime endpoint
val runtime = OpenAIServiceFactory.forBedrock(endpoint = BedrockEndpoint.Runtime)
runtime.createChatCompletion(Seq(UserMessage("Hi")),
  CreateChatCompletionSettings(model = "global.openai.gpt-5.6-luna"))

You can use the Responses API using the familiar messages and CreateChatCompletionSettings pattern via the OpenAIStreamedServiceImplicits._ extension. This is useful for accessing Responses-native tools like WebSearchTool() and CodeInterpreterTool() on models that require the Responses API (like GPT-6).

You can also convert a Responses-based service into a reusable chat-completion service using .responsesAsChatCompletion.

import io.cequence.openaiscala.domain.settings.ResponsesChatCompletionSettingsOps._

val service = OpenAIServiceFactory.withStreaming()

service
    .createChatToolCompletionStreamedViaResponses(
      messages = Seq(UserMessage("Search the news about Jupiter missions, then call get_weather for Oslo.")),
      tools = Seq(weather),
      settings = CreateChatCompletionSettings(ModelId.gpt_5_5, reasoning_effort = Some(ReasoningEffort.low))
        .setResponsesTools(Seq(WebSearchTool(), CodeInterpreterTool(container = CodeInterpreterContainer.Auto())))
    )
    .runWith(Sink.foreach(println))

// Or as a reusable service
val chatViaResponses: OpenAIChatCompletionStreamedService = service.responsesAsChatCompletion

The openai-scala-typesafe-client provides access to TypeSafe's System One API (POST /v1/systemone, model jev-latest). Unlike chat models, System One is designed for high-speed (~100ms) structured extraction. You provide a state (text or JSON) and a map of typed questions. The API evaluates all questions independently in a single call and returns typed answers with calibrated probabilities.

Question Types

• ChoiceQuestion: Selects one option from a fixed set.
• ScoreQuestion: Selects a level on an ordered rubric (numeric range).
• NoulQuestion: A yes/no question.

Configuration

Requires the TYPESAFE_API_KEY environment variable. You can optionally set TYPESAFE_BASE_URL and TYPESAFE_DEFAULT_MODEL.

Limitations

• No Streaming: The API ignores the stream flag.
• Structured Output Only: The asOpenAI() adapter is strictly for structured output.
• Input Limit: Approximately 32k tokens.

import io.cequence.openaiscala.typesafe.domain._
import io.cequence.openaiscala.typesafe.service.TypeSafeServiceFactory

val typeSafe = TypeSafeServiceFactory()  // Uses TYPESAFE_API_KEY

typeSafe.systemOne(
  state = "I've been trying to connect my Stripe account for 3 days. I'm losing sales. Please help ASAP.",
  questions = Map(
    "department"  -> ChoiceQuestion("Which team should handle this", "billing" -> "Payments", "technical" -> "Bugs, integrations", "sales" -> "Pricing"),
    "frustration" -> ScoreQuestion("How frustrated is the customer?", "Calm", "Frustrated but civil", "Very angry"),
    "is_urgent"   -> NoulQuestion("The message conveys urgency")
  )
).map { response =>
  val department = response.choice("department")   // returns .choice, .confidence, .probabilities
  val frustration = response.score("frustration")  // returns .score (e.g. 1.04), .legend, .probabilities
  val urgent = response.noul("is_urgent")          // returns .noul (e.g. 0.999)
  
  if (department.confidence < 0.7) escalateToHuman() else routeTo(department.choice)
}