reference ~15 min
GenericOpenAI Pipe Class API

- [Overview](#overview)

Table of Contents

Overview

The GenericOpenAIPipe class provides a TPipe abstraction for OpenAI-compatible APIs. It supports three API modes — OpenAI Chat Completions, Anthropic Messages, and OpenAI Responses — and works with any provider that implements one of these specifications.

class GenericOpenAIPipe : Pipe()

Requirements: Call setApiKey() before init(), or set the GENERIC_OPENAI_API_KEY environment variable.

Example — OpenAI mode (default):

val pipe = GenericOpenAIPipe()
    .setApiKey(System.getenv("OPENAI_API_KEY"))
    .setModel("gpt-4o")
    .setSystemPrompt("You are a helpful assistant.")
    .init()

val result = pipe.execute("What is TPipe?")

Example — Anthropic mode:

val pipe = GenericOpenAIPipe()
    .setApiKey(System.getenv("ANTHROPIC_API_KEY"))
    .setBaseUrl("https://api.anthropic.com")
    .setModel("claude-3-5-sonnet-20241022")
    .setApiMode(ApiMode.Anthropic)
    .init()

Example — OpenAI Responses mode:

val pipe = GenericOpenAIPipe()
    .setApiKey(System.getenv("OPENAI_API_KEY"))
    .setModel("gpt-4o-2025-04-16")
    .setApiMode(ApiMode.OpenAIResponses)
    .init()

ApiMode

ApiMode is a sealed class that selects which wire format, endpoint, request serializer, response parser, SSE parser, and auth header set the pipe uses. The default is ApiMode.OpenAI.

sealed class ApiMode
{
    data object OpenAI : ApiMode()
    data object Anthropic : ApiMode()
    data object OpenAIResponses : ApiMode()
    companion object { val DEFAULT: ApiMode = OpenAI }
}
ModeEndpointAuth Header
ApiMode.OpenAI (default)${baseUrl}/chat/completionsAuthorization: Bearer <key>
ApiMode.Anthropic${baseUrl}/anthropic/v1/messagesx-api-key: <key>, anthropic-version: 2023-06-01
ApiMode.OpenAIResponses${baseUrl}/responsesAuthorization: Bearer <key>

⚠ apiMode is locked after the first API call. Once sendRequest(...) runs (i.e. after the first execute() / generateText() / generateContent()), the internal apiModeLocked flag flips to true. Calling setApiMode(...) afterwards throws IllegalStateException("apiMode cannot be changed after the first API request"). Set the mode up front, before the first call, and create a new pipe instance if you need a different mode.

OpenAIResponses is a sealed data object that targets OpenAI’s newer Responses wire spec — top-level instructions, input items, and a streaming protocol driven by response.created / response.output_text.delta / response.completed events. See OpenAIResponsesRequestSerializer / OpenAIResponsesSseParser for the wire details.

Public Functions

Authentication & Endpoint

setApiKey(key: String): GenericOpenAIPipe

Sets the API key. Required unless GENERIC_OPENAI_API_KEY environment variable is set. The pipe-level value is checked first; if blank, init() falls back to GenericOpenAIEnv.resolveApiKey().

setBaseUrl(url: String): GenericOpenAIPipe

Sets the base URL. Defaults to https://api.openai.com/v1. Must use HTTPS — passing an http:// URL throws IllegalArgumentException("baseUrl must use HTTPS for security"). Trailing slashes are stripped.

Example:

.setBaseUrl("https://api.openai.com/v1")       // OpenAI (default)
.setBaseUrl("https://api.anthropic.com")        // Anthropic
.setBaseUrl("https://openai.myenterprise.com") // Third-party proxy

Bedrock Mantle

Amazon Bedrock Mantle is a regional Bedrock endpoint surface that exposes selected foundation models over an OpenAI-compatible HTTP wire format. The Mantle endpoint pattern is https://bedrock-mantle.{region}.api.aws/openai/v1. Mantle models are reached through GenericOpenAIPipe, not through BedrockPipe — Mantle does not speak the AWS Converse API.

Three builders configure Mantle on a pipe. The first two set the endpoint, region, and API mode; the third overrides the authentication shape that the pipe resolves from environment variables.

setBedrockMantle(region: String, modelId: String): GenericOpenAIPipe

Wires the pipe to the Mantle endpoint at https://bedrock-mantle.{region}.api.aws/openai/v1, sets apiMode to ApiMode.OpenAI, and resolves authentication. region is the AWS region code (e.g. us-east-2); modelId is the Bedrock model identifier (e.g. google.gemma-4-31b).

Example — Bearer auth via env var:

val pipe = GenericOpenAIPipe()
    .setBedrockMantle(region = "us-east-2", modelId = "google.gemma-4-31b")
    .setMaxTokens(64)
    .init()

Example — SigV4 auth via standard AWS env vars (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY):

val pipe = GenericOpenAIPipe()
    .setBedrockMantle(region = "us-east-2", modelId = "google.gemma-4-31b")
    .init()
// pipe automatically picks SigV4 from AWS_ACCESS_KEY_ID + AWS_SECRET_ACCESS_KEY

setBedrockMantleWithResponses(region: String, modelId: String): GenericOpenAIPipe

Mirrors setBedrockMantle but selects ApiMode.OpenAIResponses. Requests dispatch to ${baseUrl}/responses and the parser accepts both OpenAI’s response.reasoning_text.delta and Mantle’s shorter response.reasoning.delta SSE event names.

Example — Responses API with reasoning:

val pipe = GenericOpenAIPipe()
    .setBedrockMantleWithResponses(region = "us-east-2", modelId = "google.gemma-4-31b")
    .setMaxTokens(8192)
    .setReasoningConfig(ReasoningConfig(effort = "high"))
    .init()

setBedrockMantleAuth(auth: BedrockMantleAuth?): GenericOpenAIPipe

Replaces the Mantle authentication shape that the pipe resolves from BedrockMantleEnv. Pass null to clear any previously-set Mantle auth and fall back to the bearer/x-api-key defaults produced by the generic getAuthHeaders path. The BedrockMantleAuth sealed class has three variants:

  • BedrockMantleAuth.Bearer(apiKey) — Bearer token authentication using a Bedrock API key. Sends Authorization: Bearer <apiKey>.
  • BedrockMantleAuth.SigV4(signer) — AWS SigV4 for non-streaming requests.
  • BedrockMantleAuth.Streaming(initialSigner, chunkedSigner, decodedContentLength) — AWS SigV4 chunked-encoding for streaming requests.

The pipe picks Streaming automatically when streaming is enabled and SigV4 credentials are resolvable; it picks SigV4 for non-streaming when SigV4 credentials are resolvable; otherwise it falls back to Bearer. Explicit setBedrockMantleAuth(...) overrides the auto-resolution.

Example — explicit Bearer auth from a Bedrock API key:

val pipe = GenericOpenAIPipe()
    .setBedrockMantle(region = "us-east-2", modelId = "google.gemma-4-31b")
    .setBedrockMantleAuth(BedrockMantleAuth.bearer(System.getenv("BEDROCK_MANTLE_API_KEY")))
    .init()

IAM actions for SigV4: the AWS service identifier is bedrock-mantle. Required actions are documented in the AWS Bedrock Mantle IAM reference; the typical minimum is bedrock-mantle:CreateInference plus the standard Get* / List* actions for the model you are calling. Bearer-key users authenticate through the Bedrock API key gateway and do not require IAM permissions on the caller.

The full reference for BedrockMantleAuth (every sealed-class variant, every companion-object factory, the SigV4 streaming chunk wire format), BedrockMantleConfiguration, and BedrockMantleEnv (env-var precedence) lives in docs/api/bedrock-mantle.md. The provider-level getting-started — env-var resolution chain, model catalog, streaming and reasoning examples, error mapping — lives in docs/bedrock/mantle.md.

API Mode

setApiMode(mode: ApiMode): GenericOpenAIPipe

Sets the wire format and endpoint. Default is ApiMode.OpenAI. Throws IllegalStateException if called after the first API request.

Parameters:

  • ApiMode.OpenAI — OpenAI Chat Completions format at ${baseUrl}/chat/completions
  • ApiMode.Anthropic — Anthropic messages format at ${baseUrl}/anthropic/v1/messages
  • ApiMode.OpenAIResponses — OpenAI Responses format at ${baseUrl}/responses

Example:

.setApiMode(ApiMode.Anthropic)

Inference Settings

setFrequencyPenalty(penalty: Double): GenericOpenAIPipe

Sets frequency penalty (-2.0 to 2.0). Reduces repetition of tokens proportional to their prior frequency.

setModalities(modalities: List<String>): GenericOpenAIPipe

Sets output modalities (e.g., ["text", "image", "audio"]) for multimodal models. Serialized to the modalities field of the request body when set.

Function Calling

setTools(tools: List<ToolDefinition>): GenericOpenAIPipe

Registers function definitions for tool-calling models. See ToolDefinition / FunctionSchema for the data class shape.

Example:

import genericOpenAIPipe.env.ToolDefinition
import genericOpenAIPipe.env.FunctionSchema
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.jsonObject

val tools = listOf(
    ToolDefinition(
        type = "function",
        function = FunctionSchema(
            name = "get_weather",
            description = "Get current weather",
            parameters = Json.parseToJsonElement("""{
                "type": "object",
                "properties": {
                    "location": {"type": "string"}
                },
                "required": ["location"]
            }""").jsonObject
        )
    )
)

pipe.setTools(tools)

setToolChoice(choice: String): GenericOpenAIPipe

Sets the tool choice mode: "auto", "none", or "required".

setParallelToolCalls(enabled: Boolean): GenericOpenAIPipe

Enables or disables parallel function calling (default: true).

Structured Output

setResponseFormat(type: String, jsonSchema: kotlinx.serialization.json.JsonObject? = null): GenericOpenAIPipe

Sets the response format for structured output. Internally stores a ResponseFormat instance.

  • "text" — plain text (default)
  • "json_object" — JSON object mode
  • "json_schema" — requires jsonSchema parameter

setStructuredOutputs(enabled: Boolean): GenericOpenAIPipe

Enables structured outputs via json_schema. Must be used with setResponseFormat("json_schema", schema).

Streaming

setStreamingEnabled(enabled: Boolean): GenericOpenAIPipe

Enables Server-Sent Events (SSE) streaming. When enabled, the response is delivered as a series of chunks. The active ApiMode selects the SSE parser:

  • ApiMode.OpenAI → SseParser
  • ApiMode.Anthropic → AnthropicSseParser
  • ApiMode.OpenAIResponses → OpenAIResponsesSseParser

setStreamingCallback(callback: suspend (String) -> Unit, propagateToChildren: Boolean = true, propagateToReasoning: Boolean = true): GenericOpenAIPipe

Registers a callback for streaming response chunks. Automatically enables streaming by flipping streamingEnabled = true, adding the callback via obtainStreamingCallbackManager(), and propagating the callback to descendant pipes (validator, transformation, branch, reasoning) via propagateStreamingCallback. This ensures chunks emitted by any pipe in the tree flow through the registered callback.

Parameters:

  • callback: Suspendable callback receiving text chunks as they arrive
  • propagateToChildren: Whether to propagate to validator, transformation, and branch pipes. Defaults to true. Set to false to restrict propagation to the reasoning pipe only.
  • propagateToReasoning: Whether to propagate to the reasoning pipe. Defaults to true. Set to false to exclude the reasoning pipe.

Example — default propagation (all descendants):

pipe.setStreamingCallback { chunk ->
    print(chunk)
    flush()
}

Example — propagate to children but not to reasoning:

pipe.setStreamingCallback(
    { chunk -> print(chunk) },
    propagateToChildren = true,
    propagateToReasoning = false
)

Example — propagate to reasoning only:

pipe.setStreamingCallback(
    { chunk -> display(chunk) },
    propagateToChildren = false,
    propagateToReasoning = true
)

See Also: Streaming Callbacks Guide

Reasoning

setReasoningConfig(config: ReasoningConfig): GenericOpenAIPipe

Configures reasoning for capable models (e.g., o3, o4-mini, DeepSeek-R1). Serialized into the request body for the active mode.

data class ReasoningConfig(
    val effort: String? = null,       // "xhigh", "high", "medium", "low", "minimal", "none"
    @SerialName("max_tokens")
    val maxTokens: Int? = null,      // max tokens for reasoning output
    val exclude: Boolean? = null,    // exclude reasoning from final output
    val enabled: Boolean? = null      // enable/disable reasoning
)

Example:

import genericOpenAIPipe.env.ReasoningConfig

pipe.setReasoningConfig(ReasoningConfig(
    effort = "high",
    maxTokens = 8192,
    exclude = false,
    enabled = true
))

Prompt Caching

setCacheControl(type: String = "ephemeral", ttl: String? = null): GenericOpenAIPipe

Enables explicit prompt caching on the Anthropic API path. The cache breakpoint is placed on the last system block, caching the full system prompt prefix (tools + system) per the Anthropic/MiniMax spec.

Supported models: MiniMax-M2.7, M2.5, M2.1, M2. Not supported on M3 — use passive auto-cache on ApiMode.OpenAI instead.

TTL behavior by provider:

ProviderTTL support
Direct Anthropic API"5m" (default, 5 min) or "1h" (1 hour)
MiniMax /anthropic endpointTTL ignored — cache is always 5 minutes, auto-refreshes on hit at no additional cost

Example — MiniMax (no TTL):

val pipe = GenericOpenAIPipe()
    .setApiKey(System.getenv("MINIMAX_API_KEY"))
    .setBaseUrl("https://api.minimax.io")
    .setModel("MiniMax-M2.7")
    .setApiMode(ApiMode.Anthropic)
    .setSystemPrompt(systemPrompt)
    .setCacheControl()  // ttl omitted — MiniMax uses 5 min default
    .init()

Example — Direct Anthropic (with 1h TTL):

val pipe = GenericOpenAIPipe()
    .setApiKey(System.getenv("ANTHROPIC_API_KEY"))
    .setBaseUrl("https://api.anthropic.com")
    .setModel("claude-3-5-sonnet-20241022")
    .setApiMode(ApiMode.Anthropic)
    .setSystemPrompt(systemPrompt)
    .setCacheControl(ttl = "1h")  // 1-hour cache on Anthropic
    .init()

Note: Passive auto-cache (on ApiMode.OpenAI) requires no code — MiniMax automatically caches at 512+ input tokens at no cost. Use explicit setCacheControl only when you need the longer TTL available on direct Anthropic, or when targeting M2-family models that support it.

Resource Management

init(): Pipe

Initializes the pipe. Validates configuration and sets up the HTTP client.

Behavior:

  • Resolves API key from parameter or GenericOpenAIEnv.resolveApiKey() (which checks the env var GENERIC_OPENAI_API_KEY)
  • Sets provider = ProviderName.Gpt
  • Creates a CIO-based HTTP client with 120s request timeout, 30s connect timeout, 120s socket timeout
  • Emits a TraceEventType.PIPE_START trace event tagged provider = "GenericOpenAI"

Throws: IllegalStateException("GenericOpenAI API key is required. Call setApiKey(), genericOpenAIEnv.setApiKey(), or set GENERIC_OPENAI_API_KEY environment variable before init().") if no API key is configured.

abort()

Closes the HTTP client and cleans up resources. Emits a TraceEventType.PIPE_FAILURE trace event with action = "abort". Call when done with the pipe.

GenericOpenAIEnv

The GenericOpenAIEnv object is a process-wide singleton that owns the API key fallback chain. It lives in genericOpenAIPipe.env:

package genericOpenAIPipe.env

object GenericOpenAIEnv
{
    fun setApiKey(key: String)
    fun getApiKey(): String
    fun getApiKeyFromEnv(): String
    fun resolveApiKey(): String
    fun clearApiKey()
    fun hasApiKey(): Boolean
}

setApiKey(key: String)

Sets the Generic OpenAI API key programmatically. Persists for the lifetime of the process unless clearApiKey() is called.

import genericOpenAIPipe.env.GenericOpenAIEnv

GenericOpenAIEnv.setApiKey("sk-...")

getApiKey(): String

Returns the currently configured API key, or empty string if not set. Reads the in-process field only — does not consult the environment variable.

getApiKeyFromEnv(): String

Returns the API key from the GENERIC_OPENAI_API_KEY environment variable, or empty string if not set.

resolveApiKey(): String

Returns the effective API key: programmatic value if non-blank, otherwise the environment variable. This is the value init() uses as its fallback.

clearApiKey()

Clears the programmatically set API key. After this call, only the environment variable is used.

hasApiKey(): Boolean

Returns true if any API key is available (programmatic or env var). Useful for pre-flight checks before constructing a pipe.

Usage pattern:

import genericOpenAIPipe.env.GenericOpenAIEnv
import genericOpenAIPipe.GenericOpenAIPipe

if (!GenericOpenAIEnv.hasApiKey()) {
    GenericOpenAIEnv.setApiKey(System.getenv("OPENAI_API_KEY"))
}

val pipe = GenericOpenAIPipe()
    .setModel("gpt-4o")
    .init()

The init() chain is: pipe-level apiKey field → GenericOpenAIEnv.resolveApiKey() → IllegalStateException.

ToolDefinition / FunctionSchema

setTools(...) accepts a list of ToolDefinition data classes. These are the function-calling payloads serialized to the tools array of the request body.

package genericOpenAIPipe.env

import kotlinx.serialization.Serializable
import kotlinx.serialization.json.JsonObject

@Serializable
data class ToolDefinition(
    val type: String = "function",
    val function: FunctionSchema
)

@Serializable
data class FunctionSchema(
    val name: String,
    val description: String,
    val parameters: JsonObject
)
FieldTypeRequiredDescription
ToolDefinition.typeStringoptional, default "function"Tool type discriminator. Only "function" is supported today.
ToolDefinition.functionFunctionSchemayesThe callable function schema.
FunctionSchema.nameStringyesFunction name; must match across calls.
FunctionSchema.descriptionStringyesNatural-language description of the function. The model uses this to decide when to call it.
FunctionSchema.parametersJsonObjectyesStandard JSON Schema describing the function’s parameters.

Pair setTools(...) with setToolChoice("auto" | "none" | "required") to control when the model may call, and setParallelToolCalls(true | false) to allow multiple tool calls in a single response.

ResponseFormat

setResponseFormat(type, jsonSchema) constructs a ResponseFormat instance and stores it on the pipe. It is serialized into the response_format field of the request body.

package genericOpenAIPipe.env

import kotlinx.serialization.Serializable
import kotlinx.serialization.json.JsonObject

@Serializable
data class ResponseFormat(
    val type: String,           // "text", "json_object", or "json_schema"
    val jsonSchema: JsonObject? = null
)
FieldTypeRequiredDescription
typeStringyesFormat type. Use "text" for plain prose, "json_object" for free-form JSON, "json_schema" for schema-validated JSON.
jsonSchemaJsonObject?required when type == "json_schema"A standard JSON Schema describing the desired output shape.

For schema-validated output, also call setStructuredOutputs(true):

val schema = Json.parseToJsonElement("""{
    "type": "object",
    "properties": { "answer": { "type": "string" } },
    "required": ["answer"]
}""").jsonObject

pipe.setResponseFormat("json_schema", schema)
pipe.setStructuredOutputs(true)

CacheControl

Anthropic-style prompt caching hint carried on the wire-level request. Defined in genericOpenAIPipe.env:

package genericOpenAIPipe.env

import kotlinx.serialization.Serializable

@Serializable
data class CacheControl(
    val type: String,           // e.g., "ephemeral"
    val ttl: String? = null     // e.g., "5m", "1h", "24h"
)
FieldTypeRequiredDescription
typeStringyesCache type discriminator. Common value: "ephemeral".
ttlString?optionalCache time-to-live. Provider-accepted values include "5m", "1h", "24h".

CacheControl is a field on GenericOpenAIChatRequest (serialized as cache_control) and is plumbed through generateText(...) and generateContent(...). The pipe does not currently expose a public setCacheControl(...) builder — CacheControl is intended for direct request construction or custom request pipelines. For higher-level caching ergonomics, prefer setApiMode(ApiMode.Anthropic) and let the Anthropic provider handle prompt caching at the protocol level.

Error Mapping

GenericOpenAIPipe maps transport-level and provider-level failures to the P2PError enum (values: auth, prompt, json, content, transport, context, configuration, none):

HTTP / conditionProvider error typeP2PErrorWhen
401—authInvalid or missing API key
403—authForbidden (region / model access)
400invalid_request_errorpromptMalformed request body
400invalid_api_keypromptProvider rejected the key shape
429rate_limit_errortransportRate limit
5xxapi_error, server_errortransportProvider server error
——transportNetwork / socket / request timeout
——jsonResponse body could not be parsed into GenericOpenAIChatResponse

The mapping is enforced in two places:

  • SSE path (executeStreamingOpenAI) — deserializes the first data: payload as GenericOpenAIErrorResponse and classifies on error.type (authentication_error → auth, rate_limit_error → transport, invalid_request_error / invalid_api_key → prompt, api_error / server_error → transport, default → transport).
  • Non-streaming path — converts HttpRequestTimeoutException, SocketTimeoutException, and ConnectException into P2PException(P2PError.transport, ...), and wraps JSON parse failures as P2PException(P2PError.json, ...).

Catch P2PException and inspect errorType for typed handling.

Multimodal Content

GenericOpenAIPipe supports multimodal inputs (images, documents) via MultimodalContent.binaryContent. Binary content is automatically converted to the appropriate format for the selected ApiMode inside generateContent(...):

BinaryContent typeOpenAI modeAnthropic modeOpenAI Responses mode
Bytesbase64 data URI image blockbase64 image blockbase64 input item
Base64Stringbase64 data URI image blockbase64 image blockbase64 input item
CloudReferenceURL image blockURL image blockURL input item
TextDocumenttext blocktext blocktext input item

Bytes and Base64String are equivalent on the wire — both end up as a data:<mime>;base64,... URL inside an ImageUrlBlock. CloudReference passes the URL through as-is. TextDocument injects the text into the content array as a TextBlock.

Example:

import com.TTT.Pipe.MultimodalContent
import com.TTT.Pipe.BinaryContent
import genericOpenAIPipe.GenericOpenAIPipe
import genericOpenAIPipe.api.ApiMode
import kotlinx.coroutines.runBlocking
import java.io.File

fun main() = runBlocking {
    val imageBytes = File("paris.png").readBytes()
    val pipe = GenericOpenAIPipe()
        .setApiKey(System.getenv("OPENAI_API_KEY"))
        .setModel("gpt-4o")
        .setApiMode(ApiMode.OpenAI)
        .init()

    val content = MultimodalContent(
        text = "What is in this image?",
        binaryContent = mutableListOf(
            BinaryContent.Bytes(data = imageBytes, mimeType = "image/png")
        )
    )

    println(pipe.execute(content).text)
}

Next Steps

  • Bedrock Mantle Getting Started — Amazon Bedrock Mantle provider guide: endpoint, authentication (Bearer / SigV4 / chunked-streaming), reasoning, streaming, error conditions.
  • Bedrock Mantle API Reference — BedrockMantleConfiguration, BedrockMantleAuth (Bearer / SigV4 / Streaming variants), BedrockMantleEnv env-var precedence.
  • Getting Started with GenericOpenAI — Provider recipes, comparison with OllamaPipe / OpenRouterPipe / BedrockPipe, and a troubleshooting guide.
  • Pipe Context Protocol — Attach PCP tools to a GenericOpenAIPipe for sandboxed multi-language tool execution.
  • Pipe Class API — Core pipe abstraction and base-class builders (setModel, setTemperature, setMaxTokens, setSystemPrompt, setStopSequences, setSeed, setUser, setLogitBias, setN, setRepetitionPenalty, setPresencePenalty, setContextWindowSize, setTokenBudget).
  • DistributionGrid — Route tasks across workers with a GenericOpenAIPipe router.