Streaming chat completions in plain Swift with URLSession.bytes (no SDK)

Community Showcase
, , ,
1

Most LLM APIs now speak the same "OpenAI-style" HTTP protocol. You don't need a package for it: URLSession.bytes(for:) plus AsyncLineSequence handles the server-sent-event stream in about 40 lines. Here is the version I keep reusing (macOS 12 / iOS 15+).

import Foundation

struct ChatRequest: Encodable {
    struct Message: Encodable { let role: String; let content: String }
    let model: String
    let stream: Bool
    let messages: [Message]
}

struct ChatChunk: Decodable {
    struct Choice: Decodable {
        struct Delta: Decodable { let content: String? }
        let delta: Delta
    }
    let choices: [Choice]
}

struct HTTPError: Error { let status: Int; let body: String }

func streamChat(_ prompt: String, onToken: (String) -> Void) async throws {
    let baseURL = URL(string: ProcessInfo.processInfo.environment["LLM_BASE_URL"]
                      ?? "https://api.china-aiapi.com/v1")!
    let apiKey = ProcessInfo.processInfo.environment["LLM_API_KEY"] ?? ""

    var request = URLRequest(url: baseURL.appendingPathComponent("chat/completions"))
    request.httpMethod = "POST"
    request.setValue("Bearer \(apiKey)", forHTTPHeaderField: "Authorization")
    request.setValue("application/json", forHTTPHeaderField: "Content-Type")
    request.timeoutInterval = 60   // idle time between bytes, not total duration
    request.httpBody = try JSONEncoder().encode(
        ChatRequest(model: "MODEL_ID", stream: true,
                    messages: [.init(role: "user", content: prompt)]))

    let (bytes, response) = try await URLSession.shared.bytes(for: request)
    let status = (response as? HTTPURLResponse)?.statusCode ?? 0
    guard status == 200 else {
        var body = ""
        for try await line in bytes.lines { body += line }
        throw HTTPError(status: status, body: body)
    }

    let decoder = JSONDecoder()
    for try await line in bytes.lines {
        guard line.hasPrefix("data:") else { continue }   // skips ": keep-alive" comments
        let payload = line.dropFirst(5).trimmingCharacters(in: .whitespaces)
        if payload == "[DONE]" { break }
        let chunk = try decoder.decode(ChatChunk.self, from: Data(payload.utf8))
        if let token = chunk.choices.first?.delta.content { onToken(token) }
    }
}

Usage, with cancellation:

let task = Task {
    try await streamChat("Explain Sendable in two sentences.") { print($0, terminator: "") }
}
// task.cancel() stops the loop and closes the connection

Things that bit me:

  • timeoutInterval is the idle timeout. A long answer that keeps streaming will never hit it, but a slow first token on a big model can. 60 s is a safer start than the 30 s that is common in examples.
  • On a non-200 status the body is plain JSON, not SSE. Read it before throwing, otherwise you only see "400".
  • The last chunk can have an empty choices array (for example when usage reporting is on), hence choices.first?.
  • bytes(for:) is Apple-platforms only. swift-corelibs-foundation on Linux doesn't have AsyncBytes, so server-side Swift needs AsyncHTTPClient instead.
  • Keep the key out of the app bundle for anything you ship. Read it from the environment in tools, and use your own backend for apps.

Switching providers is just LLM_BASE_URL plus the model ID. Disclosure: I work on the APIGOTO team; APIGOTO is independent of OpenAI, Anthropic and Google. The default URL above is ours (www.china-aiapi.com), but the code works with any compatible endpoint, including a local server.