Documentation

AI providers

Kynetra FX ships three production AiPort adapters:openAiAdapter for OpenAI-compatible APIs, anthropicAdapter for Anthropic Claude, and workersAiAdapter for Cloudflare Workers AI. All three accept an injectable fetch function so you can test them in isolation or proxy requests through a gateway.

openAiAdapter#

openAiAdapter talks to the OpenAI Chat Completions and Embeddings APIs. Set baseUrl to point at any OpenAI-compatible endpoint — Azure OpenAI, Together AI, Groq, or a local Ollama instance.

Name	Type	Description
apiKey	string	OpenAI API key (required).
baseUrl	string?	Base URL for the API. Defaults to https://api.openai.com/v1.
model	string?	Default chat model. Defaults to gpt-4o-mini. Can be overridden per call via the model field on complete() / embed().
fetch	typeof globalThis.fetch?	Custom fetch implementation. Defaults to globalThis.fetch. Pass a test stub to intercept HTTP calls.

src/app.ts

import { createFX } from '@kynetra/fx'
import { openAiAdapter } from '@kynetra/fx-ai'
 
const app = createFX({ name: 'my-api' })
 
app.decorateContext('ai', openAiAdapter({
  apiKey: process.env.OPENAI_API_KEY!,
  model: 'gpt-4o',
}))
 
app.post('/summarise', async (ctx) => {
  const ai = ctx.get('ai')
  const { text } = await ctx.jsonBody<{ text: string }>()
 
  const { text: summary } = await ai.complete({
    messages: [
      { role: 'system', content: 'Summarise in one paragraph.' },
      { role: 'user', content: text },
    ],
  })
 
  return ctx.json({ summary })
})

Per-call model override

Pass model inside the complete() or embed() call to override the adapter default for that specific request:

examples/model-override.ts

const ai = openAiAdapter({ apiKey: '...', model: 'gpt-4o-mini' })
 
// Use a stronger model for a critical path
const { text } = await ai.complete({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Explain quantum entanglement.' }],
})
 
// Use a dedicated embedding model
const { vectors } = await ai.embed({
  model: 'text-embedding-3-large',
  input: ['hello world'],
})

OpenAI-compatible endpoints

examples/groq.ts

import { openAiAdapter } from '@kynetra/fx-ai'
 
// Point at Groq's OpenAI-compatible API
const ai = openAiAdapter({
  apiKey: process.env.GROQ_API_KEY!,
  baseUrl: 'https://api.groq.com/openai/v1',
  model: 'llama-3.1-8b-instant',
})

anthropicAdapter#

anthropicAdapter calls the Anthropic Messages API. It maps the generic complete() arguments onto Anthropic's messages format, extracting system role messages automatically.

Name	Type	Description
apiKey	string	Anthropic API key (required).
baseUrl	string?	Override the Anthropic API base URL. Defaults to https://api.anthropic.com.
model	string?	Default model. Defaults to claude-3-5-haiku-20241022.
fetch	typeof globalThis.fetch?	Injectable fetch for testing or proxying.

src/app-anthropic.ts

import { createFX } from '@kynetra/fx'
import { anthropicAdapter } from '@kynetra/fx-ai'
 
const app = createFX()
 
app.decorateContext('ai', anthropicAdapter({
  apiKey: process.env.ANTHROPIC_API_KEY!,
  model: 'claude-opus-4-5',   // use a stronger model for complex tasks
}))
 
app.post('/review', async (ctx) => {
  const ai = ctx.get('ai')
  const { code } = await ctx.jsonBody<{ code: string }>()
 
  const { text: review } = await ai.complete({
    messages: [
      { role: 'system', content: 'You are an expert TypeScript code reviewer.' },
      { role: 'user', content: `Review this code for bugs and improvements:\n```ts\n${code}\n```` },
    ],
  })
 
  return ctx.json({ review })
})

Note

Anthropic's API does not support text embeddings natively. Calling embed() on anthropicAdapter throws — use openAiAdapter or workersAiAdapter for the embedding step, and anthropicAdapter for completion.

workersAiAdapter#

workersAiAdapter uses a Cloudflare Workers AI binding directly — no API key, no outbound HTTP. Inference runs on Cloudflare's GPU network at the edge, billed per token.

Name	Type	Description
binding	Ai	The env.AI Workers AI binding from your wrangler.toml (required).
model	string?	Default model ID. Defaults to @cf/meta/llama-3.1-8b-instruct for complete() and @cf/baai/bge-small-en-v1.5 for embed().

src/worker.ts

import { cloudflare } from '@kynetra/fx-cloudflare'
import { workersAiAdapter } from '@kynetra/fx-ai'
import app from './app'
 
interface Env {
  AI: Ai
}
 
return {
  fetch(request: Request, env: Env, ctx: ExecutionContext) {
    // Bind the Workers AI adapter at the edge — no API key needed
    app.decorateContext('ai', workersAiAdapter({ binding: env.AI }))
    return cloudflare(app).fetch(request, env, ctx)
  }
}

wrangler.toml

name = "my-worker"
main = "src/worker.ts"
compatibility_date = "2024-11-01"
 
[ai]
binding = "AI"

Injectable fetch — testing adapters in isolation#

Every HTTP-backed adapter accepts a fetch option. Pass a stub to intercept and assert on outbound API calls without hitting the network.

src/ai.test.ts

import { describe, it, expect } from 'vitest'
import { openAiAdapter } from '@kynetra/fx-ai'
 
it('sends messages to the OpenAI API', async () => {
  let capturedBody: unknown
 
  const fakeFetch: typeof fetch = async (input, init) => {
    capturedBody = JSON.parse(init?.body as string)
    return new Response(JSON.stringify({
      choices: [{ message: { content: 'Paris' } }],
    }), { headers: { 'content-type': 'application/json' } })
  }
 
  const ai = openAiAdapter({ apiKey: 'test-key', fetch: fakeFetch })
  const { text } = await ai.complete({
    messages: [{ role: 'user', content: 'Capital of France?' }],
  })
 
  expect(text).toBe('Paris')
  expect((capturedBody as { messages: unknown[] }).messages).toHaveLength(1)
})

Error handling#

All adapters reject with a native Error if the upstream API returns a non-2xx response. The error message includes the HTTP status code and the response body when available. Wrap calls in a try/catch and map provider errors to your own error types at the service boundary:

src/services/ai-service.ts

import type { AiPort } from '@kynetra/fx'
import { FXError } from '@kynetra/fx'
 
export async function safeComplete(
  ai: AiPort,
  prompt: string,
): Promise<string> {
  try {
    const { text } = await ai.complete({
      messages: [{ role: 'user', content: prompt }],
    })
    return text
  } catch (err) {
    // Map upstream errors to a 503 so callers get a consistent shape
    throw new FXError(503, 'FX_AI_UNAVAILABLE', 'AI provider error', {
      cause: String(err),
    })
  }
}

Choosing a provider#

Development / CI — use mockAi. Zero cost, zero latency, deterministic.
Cloudflare Workers — workersAiAdapter for zero-config inference; openAiAdapter or anthropicAdapter when you need a specific frontier model.
Node / Bun / Deno — openAiAdapter or anthropicAdapter depending on model preference.
OpenAI-compatible providers (Groq, Together AI, Azure, Ollama) — openAiAdapter with a custom baseUrl.
Embeddings — openAiAdapter (text-embedding-3-small / text-embedding-3-large) or workersAiAdapter (@cf/baai/bge-small-en-v1.5).

PreviousAI port NextRAG