Repository: softwaremill/sttp-openai
Pull request #124 (2023-11-06 10:35:52)

Add streaming support for chat completions
Closes https://github.com/softwaremill/sttp-openai/issues/59

This PR adds streaming support for [Create chat completion](https://platform.openai.com/docs/api-reference/chat/create) endpoint using server-sent events.

_**Note**_: For the first version we used `Fs2` streams, we'll support other stream implementations in the feature.

Here are the main changes and enhancements made in this PR:

- case classes related to chat completions have been redefined to align with the latest OpenAI API. This ensures compatibility and consistency with the OpenAI API.
- `OpenAI` class has been expanded with new `createChatCompletion` method. This method is designed to work with generic stream implementations using `Streams[S]`
- **out-of-the-box** streaming support has been added using the `Fs2` library by extending current `OpenAI` class using implicit class:
```scala
package sttp.openai.streaming.fs2

object Fs2OpenAI {
  implicit class streaming(val client: OpenAI) {
    def createStreamedChatCompletion[F[_]: RaiseThrowable](chatBody: ChatBody): StreamRequest[Either[OpenAIException, Stream[F, ChatChunkResponse]], Fs2Streams[F]]
  }
}
```
- README has been updated to include an example demonstrating how to use this new feature
- integration tests have been added using `HttpClientFs2Backend.stub`


Patch:
diff --git a/README.md b/README.md
index 974b4d4..1a32e22 100644
--- a/README.md
+++ b/README.md
@@ -28,11 +28,6 @@ sttp openai is available for Scala 2.13 and Scala 3
 
 OpenAI API Official Documentation https://platform.openai.com/docs/api-reference/completions
 
-### Not yet implemented:
-    * Create chat completions SSE
-    * Create completions SSE
-    * List fine-tune events SSE
-
 ## Example
 
 ### To use ChatGPT
@@ -40,9 +35,8 @@ OpenAI API Official Documentation https://platform.openai.com/docs/api-reference
 ```scala mdoc:compile-only 
 import sttp.openai.OpenAISyncClient
 import sttp.openai.requests.completions.chat.ChatRequestResponseData.ChatResponse
-import sttp.openai.requests.completions.chat.ChatRequestBody.{ChatBody, ChatCompletionModel}
-import sttp.openai.requests.completions.chat.{Message, Role}
-
+import sttp.openai.requests.completions.chat.ChatRequestBody.{ChatBody, ChatCompletionModel, Message}
+import sttp.openai.requests.completions.chat.Role
 
 object Main extends App {
   // Create an instance of OpenAISyncClient providing your API secret-key
@@ -92,11 +86,12 @@ or use backend of your choice.
 ```scala mdoc:compile-only 
 import cats.effect.{ExitCode, IO, IOApp}
 import sttp.client4.httpclient.cats.HttpClientCatsBackend
+
+import sttp.openai.OpenAI
 import sttp.openai.OpenAIExceptions.OpenAIException
-import sttp.openai._
-import sttp.openai.requests.completions.chat.ChatRequestBody.{ChatBody, ChatCompletionModel}
 import sttp.openai.requests.completions.chat.ChatRequestResponseData.ChatResponse
-import sttp.openai.requests.completions.chat.{Message, Role}
+import sttp.openai.requests.completions.chat.ChatRequestBody.{ChatBody, ChatCompletionModel, Message}
+import sttp.openai.requests.completions.chat.Role
 
 object Main extends IOApp {
   override def run(args: List[String]): IO[ExitCode] = {
@@ -143,6 +138,95 @@ object Main extends IOApp {
   */
 }
 ```
+
+#### Create completion with streaming:
+The Chat Completion API features streaming support via server-sent events. Currently, we only support streaming using `Fs2`. 
+
+Add the following import:
+
+```scala
+import sttp.openai.streaming.fs2._
+```
+
+Example below uses `HttpClientFs2Backend` as a backend.
+
+```scala mdoc:compile-only
+import cats.effect.{ExitCode, IO, IOApp}
+import fs2.Stream
+import sttp.client4.httpclient.fs2.HttpClientFs2Backend
+
+import sttp.openai.OpenAI
+import sttp.openai.streaming.fs2._
+import sttp.openai.OpenAIExceptions.OpenAIException
+import sttp.openai.requests.completions.chat.ChatChunkRequestResponseData.ChatChunkResponse
+import sttp.openai.requests.completions.chat.ChatRequestBody.{ChatBody, ChatCompletionModel, Message}
+import sttp.openai.requests.completions.chat.Role
+
+object Main extends IOApp {
+  override def run(args: List[String]): IO[ExitCode] = {
+    val openAI: OpenAI = new OpenAI("your-secret-key")
+
+    val bodyMessages: Seq[Message] = Seq(
+      Message(
+        role = Role.User,
+        content = "Hello!"
+      )
+    )
+
+    val chatRequestBody: ChatBody = ChatBody(
+      model = ChatCompletionModel.GPT35Turbo,
+      messages = bodyMessages
+    )
+
+    HttpClientFs2Backend.resource[IO]().use { backend =>
+      val response: IO[Either[OpenAIException, Stream[IO, ChatChunkResponse]]] =
+        openAI
+          .createStreamedChatCompletion[IO](chatRequestBody)
+          .send(backend)
+          .map(_.body)
+
+      response
+        .flatMap {
+          case Left(exception) => IO.println(exception.getMessage)
+          case Right(stream)   => stream.evalTap(IO.println).compile.drain
+        }
+        .as(ExitCode.Success)
+    }
+  }
+  /*
+    ...
+    ChatChunkResponse(
+      "chatcmpl-8HEZFNDmu2AYW8jVvNKyRO4W4KcO8",
+      "chat.completion.chunk",
+      1699118265,
+      "gpt-3.5-turbo-0613",
+      List(
+        Choices(
+          Delta(None, Some("Hi"), None),
+          null,
+          0
+        )
+      )
+    )
+    ...
+    ChatChunkResponse(
+      "chatcmpl-8HEZFNDmu2AYW8jVvNKyRO4W4KcO8",
+      "chat.completion.chunk",
+      1699118265,
+      "gpt-3.5-turbo-0613",
+      List(
+        Choices(
+          Delta(None, Some(" there"), None),
+          null,
+          0
+        )
+      )
+    )
+    ...
+   */
+}
+```
+
 ## Contributing
 
 If you have a question, or hit a problem, feel free to post on our community https://softwaremill.community/c/open-source/
diff --git a/build.sbt b/build.sbt
index 8d007f0..7e1b053 100644
--- a/build.sbt
+++ b/build.sbt
@@ -15,7 +15,11 @@ lazy val commonSettings = commonSmlBuildSettings ++ ossPublishSettings ++ Seq(
 lazy val root = (project in file("."))
   .settings(commonSettings: _*)
   .settings(publish / skip := true, name := "sttp-openai", scalaVersion := scala2.head)
-  .aggregate(core.projectRefs: _*)
+  .aggregate(allAgregates: _*)
+
+lazy val allAgregates = core.projectRefs ++
+  fs2.projectRefs ++
+  docs.projectRefs
 
 lazy val core = (projectMatrix in file("core"))
   .jvmPlatform(
@@ -26,6 +30,16 @@ lazy val core = (projectMatrix in file("core"))
   )
   .settings(commonSettings: _*)
 
+lazy val fs2 = (projectMatrix in file("streaming/fs2"))
+  .jvmPlatform(
+    scalaVersions = scala2 ++ scala3
+  )
+  .settings(commonSettings)
+  .settings(
+    libraryDependencies ++= Libraries.sttpClientFs2
+  )
+  .dependsOn(core % "compile->compile;test->test")
+
 val compileDocs: TaskKey[Unit] = taskKey[Unit]("Compiles docs module throwing away its output")
 compileDocs := {
   (docs.jvm(scala2.head) / mdoc).toTask(" --out target/sttp-openai-docs").value
@@ -48,4 +62,5 @@ lazy val docs = (projectMatrix in file("generated-docs")) // important: it must
     evictionErrorLevel := Level.Info
   )
   .dependsOn(core)
+  .dependsOn(fs2)
   .jvmPlatform(scalaVersions = scala2)
diff --git a/core/src/main/scala/sttp/openai/OpenAI.scala b/core/src/main/scala/sttp/openai/OpenAI.scala
index c26d688..7fb6bec 100644
--- a/core/src/main/scala/sttp/openai/OpenAI.scala
+++ b/core/src/main/scala/sttp/openai/OpenAI.scala
@@ -3,7 +3,7 @@ package sttp.openai
 import sttp.client4._
 import sttp.model.Uri
 import sttp.openai.OpenAIExceptions.OpenAIException
-import sttp.openai.json.SttpUpickleApiExtension.{asJsonSnake, asStringEither, upickleBodySerializer}
+import sttp.openai.json.SttpUpickleApiExtension.{asJsonSnake, asStreamSnake, asStringEither, upickleBodySerializer}
 import sttp.openai.requests.completions.CompletionsRequestBody.CompletionsBody
 import sttp.openai.requests.completions.CompletionsResponseData.CompletionsResponse
 import sttp.openai.requests.completions.chat.ChatRequestBody.ChatBody
@@ -31,6 +31,7 @@ import sttp.openai.requests.audio.AudioResponseData.AudioResponse
 import sttp.openai.requests.audio.transcriptions.TranscriptionConfig
 import sttp.openai.requests.audio.translations.TranslationConfig
 import sttp.openai.requests.audio.RecognitionModel
+import sttp.capabilities.Streams
 
 import java.io.File
 import java.nio.file.Paths
@@ -239,6 +240,21 @@ class OpenAI(authToken: String) {
       .body(chatBody)
       .response(asJsonSnake[ChatResponse])
 
+  /** Creates and streams a model response as chunk objects for the given chat conversation defined in chatBody.
+    *
+    * [[https://platform.openai.com/docs/api-reference/chat/create]]
+    *
+    * @param s
+    *   The streams implementation to use.
+    * @param chatBody
+    *   Chat request body.
+    */
+  def createChatCompletion[S](s: Streams[S], chatBody: ChatBody): StreamRequest[Either[OpenAIException, s.BinaryStream], S] =
+    openAIAuthRequest
+      .post(OpenAIUris.ChatCompletions)
+      .body(ChatBody.withStreaming(chatBody))
+      .response(asStreamSnake(s))
+
   /** Returns a list of files that belong to the user's organization.
     *
     * [[https://platform.openai.com/docs/api-reference/files]]
diff --git a/core/src/main/scala/sttp/openai/json/SttpUpickleApiExtension.scala b/core/src/main/scala/sttp/openai/json/SttpUpickleApiExtension.scala
index d09055b..518ac79 100644
--- a/core/src/main/scala/sttp/openai/json/SttpUpickleApiExtension.scala
+++ b/core/src/main/scala/sttp/openai/json/SttpUpickleApiExtension.scala
@@ -7,6 +7,7 @@ import sttp.model.StatusCode._
 import sttp.model.ResponseMetadata
 import sttp.openai.OpenAIExceptions.OpenAIException
 import sttp.openai.OpenAIExceptions.OpenAIException._
+import sttp.capabilities.Streams
 
 /** An sttp upickle api extension that deserializes JSON with snake_case keys into case classes with fields corresponding to keys in
   * camelCase and maps errors to OpenAIException subclasses.
@@ -14,6 +15,11 @@ import sttp.openai.OpenAIExceptions.OpenAIException._
 object SttpUpickleApiExtension extends SttpUpickleApi {
   override val upickleApi: SnakePickle.type = SnakePickle
 
+  def asStreamSnake[S](s: Streams[S]): StreamResponseAs[Either[OpenAIException, s.BinaryStream], S] =
+    asStreamUnsafe(s).mapWithMetadata { (body, meta) =>
+      body.left.map(errorBody => httpToOpenAIError(HttpError(errorBody, meta.code)))
+    }
+
   def asJsonSnake[B: upickleApi.Reader: IsOption]: ResponseAs[Either[OpenAIException, B]] =
     asString.mapWithMetadata(deserializeRightWithMappedExceptions(deserializeJsonSnake)).showAsJson
 
diff --git a/core/src/main/scala/sttp/openai/requests/completions/chat/ChatChunkRequestResponseData.scala b/core/src/main/scala/sttp/openai/requests/completions/chat/ChatChunkRequestResponseData.scala
new file mode 100644
index 0000000..55ddd62
--- /dev/null
+++ b/core/src/main/scala/sttp/openai/requests/completions/chat/ChatChunkRequestResponseData.scala
@@ -0,0 +1,44 @@
+package sttp.openai.requests.completions.chat
+
+import sttp.openai.json.SnakePickle
+
+object ChatChunkRequestResponseData {
+
+  /** @param role
+    *   The role of the author of this message.
+    * @param content
+    *   The contents of the message.
+    * @param functionCall
+    *   The name of the author of this message. May contain a-z, A-Z, 0-9, and underscores, with a maximum length of 64 characters.
+    */
+  case class Delta(role: Option[Role] = None, content: Option[String] = None, functionCall: Option[FunctionCall] = None)
+
+  object Delta {
+    implicit val deltaR: SnakePickle.Reader[Delta] = SnakePickle.macroR[Delta]
+  }
+
+  case class Choices(
+      delta: Delta,
+      finishReason: String,
+      index: Int
+  )
+
+  object Choices {
+    implicit val choicesR: SnakePickle.Reader[Choices] = SnakePickle.macroR[Choices]
+  }
+
+  case class ChatChunkResponse(
+      id: String,
+      `object`: String,
+      created: Int,
+      model: String,
+      choices: Seq[Choices]
+  )
+
+  object ChatChunkResponse {
+    val DoneEventMessage = "[DONE]"
+
+    implicit val chunkChatR: SnakePickle.Reader[ChatChunkResponse] = SnakePickle.macroR[ChatChunkResponse]
+  }
+
+}
diff --git a/core/src/main/scala/sttp/openai/requests/completions/chat/ChatRequestBody.scala b/core/src/main/scala/sttp/openai/requests/completions/chat/ChatRequestBody.scala
index 6040cdd..5ff1c53 100644
--- a/core/src/main/scala/sttp/openai/requests/completions/chat/ChatRequestBody.scala
+++ b/core/src/main/scala/sttp/openai/requests/completions/chat/ChatRequestBody.scala
@@ -7,6 +7,21 @@ import ujson.Str
 
 object ChatRequestBody {
 
+  /** @param role
+    *   The role of the author of this message.
+    * @param content
+    *   The contents of the message.
+    * @param name
+    *   The name of the author of this message. May contain a-z, A-Z, 0-9, and underscores, with a maximum length of 64 characters.
+    * @param functionCall
+    *   The name and arguments of a function that should be called, as generated by the model.
+    */
+  case class Message(role: Role, content: String, name: Option[String] = None, functionCall: Option[FunctionCall] = None)
+
+  object Message {
+    implicit val messageRW: SnakePickle.ReadWriter[Message] = SnakePickle.macroRW[Message]
+  }
+
   /** @param model
     *   ID of the model to use.
     * @param messages
@@ -49,6 +64,12 @@ object ChatRequestBody {
   )
 
   object ChatBody {
+    def withStreaming(chatBody: ChatBody): ujson.Value = {
+      val json = SnakePickle.writeJs(chatBody)
+      json.obj("stream") = true
+      json
+    }
+
     implicit val chatRequestW: SnakePickle.Writer[ChatBody] = SnakePickle.macroW[ChatBody]
   }
 
diff --git a/core/src/main/scala/sttp/openai/requests/completions/chat/ChatRequestResponseData.scala b/core/src/main/scala/sttp/openai/requests/completions/chat/ChatRequestResponseData.scala
index d361361..b5021f0 100644
--- a/core/src/main/scala/sttp/openai/requests/completions/chat/ChatRequestResponseData.scala
+++ b/core/src/main/scala/sttp/openai/requests/completions/chat/ChatRequestResponseData.scala
@@ -4,6 +4,20 @@ import sttp.openai.json.SnakePickle
 import sttp.openai.requests.completions.Usage
 
 object ChatRequestResponseData {
+
+  /** @param role
+    *   The role of the author of this message.
+    * @param content
+    *   The contents of the message.
+    * @param functionCall
+    *   The name and arguments of a function that should be called, as generated by the model.
+    */
+  case class Message(role: Role, content: String, functionCall: Option[FunctionCall] = None)
+
+  object Message {
+    implicit val messageRW: SnakePickle.Reader[Message] = SnakePickle.macroR[Message]
+  }
+
   case class Choices(
       message: Message,
       finishReason: String,
diff --git a/core/src/main/scala/sttp/openai/requests/completions/chat/FunctionCall.scala b/core/src/main/scala/sttp/openai/requests/completions/chat/FunctionCall.scala
new file mode 100644
index 0000000..b988dd2
--- /dev/null
+++ b/core/src/main/scala/sttp/openai/requests/completions/chat/FunctionCall.scala
@@ -0,0 +1,16 @@
+package sttp.openai.requests.completions.chat
+
+import sttp.openai.json.SnakePickle
+
+/** @param arguments
+  *   The arguments to call the function with, as generated by the model in JSON format. Note that the model does not always generate valid
+  *   JSON, and may hallucinate parameters not defined by your function schema. Validate the arguments in your code before calling your
+  *   function.
+  * @param name
+  *   The name of the function to call.
+  */
+case class FunctionCall(arguments: String, name: String)
+
+object FunctionCall {
+  implicit val functionCallRW: SnakePickle.ReadWriter[FunctionCall] = SnakePickle.macroRW[FunctionCall]
+}
diff --git a/core/src/main/scala/sttp/openai/requests/completions/chat/Message.scala b/core/src/main/scala/sttp/openai/requests/completions/chat/Role.scala
similarity index 68%
rename from core/src/main/scala/sttp/openai/requests/completions/chat/Message.scala
rename to core/src/main/scala/sttp/openai/requests/completions/chat/Role.scala
index 50633bd..bd9a156 100644
--- a/core/src/main/scala/sttp/openai/requests/completions/chat/Message.scala
+++ b/core/src/main/scala/sttp/openai/requests/completions/chat/Role.scala
@@ -4,20 +4,8 @@ import sttp.openai.OpenAIExceptions.OpenAIException.DeserializationOpenAIExcepti
 import sttp.openai.json.SnakePickle
 import ujson.Str
 
-/** @param role
-  *   The role of the author of this message. One of [[Role]].
-  * @param content
-  *   The contents of the message.
-  * @param name
-  *   The name of the author of this message. May contain a-z, A-Z, 0-9, and underscores, with a maximum length of 64 characters.
-  */
-case class Message(role: Role, content: String, name: Option[String] = None)
-
-object Message {
-  implicit val messageRW: SnakePickle.ReadWriter[Message] = SnakePickle.macroRW[Message]
-}
-
 sealed abstract class Role(val value: String)
+
 object Role {
   case object System extends Role("system")
 
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index daf0352..1cba935 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -4,6 +4,8 @@ object Dependencies {
 
   object V {
     val scalaTest = "3.2.17"
+    val scalaTestCats = "1.5.0"
+
     val sttpClient = "4.0.0-M6"
     val uPickle = "3.1.3"
   }
@@ -17,6 +19,11 @@ object Dependencies {
       "com.softwaremill.sttp.client4" %% "upickle" % V.sttpClient
     )
 
+    val sttpClientFs2 = Seq(
+      "com.softwaremill.sttp.client4" %% "fs2" % V.sttpClient,
+      "org.typelevel" %% "cats-effect-testing-scalatest" % V.scalaTestCats % Test
+    )
+
     val uPickle = "com.lihaoyi" %% "upickle" % V.uPickle
 
   }
diff --git a/streaming/fs2/src/main/scala/sttp/openai/streaming/Fs2OpenAI.scala b/streaming/fs2/src/main/scala/sttp/openai/streaming/Fs2OpenAI.scala
new file mode 100644
index 0000000..3dc1f81
--- /dev/null
+++ b/streaming/fs2/src/main/scala/sttp/openai/streaming/Fs2OpenAI.scala
@@ -0,0 +1,48 @@
+package sttp.openai.streaming
+
+import _root_.fs2.{Pipe, RaiseThrowable, Stream}
+import sttp.capabilities.fs2.Fs2Streams
+import sttp.client4.StreamRequest
+import sttp.client4.impl.fs2.Fs2ServerSentEvents
+import sttp.model.sse.ServerSentEvent
+import sttp.openai.OpenAI
+import sttp.openai.OpenAIExceptions.OpenAIException
+import sttp.openai.json.SttpUpickleApiExtension.deserializeJsonSnake
+import sttp.openai.requests.completions.chat.ChatChunkRequestResponseData.ChatChunkResponse
+import sttp.openai.requests.completions.chat.ChatRequestBody.ChatBody
+
+package object fs2 {
+  import ChatChunkResponse.DoneEventMessage
+
+  implicit class extension(val client: OpenAI) {
+
+    /** Creates and streams a model response as chunk objects for the given chat conversation defined in chatBody.
+      *
+      * [[https://platform.openai.com/docs/api-reference/chat/create]]
+      *
+      * @param chatBody
+      *   Chat request body.
+      */
+    def createStreamedChatCompletion[F[_]: RaiseThrowable](
+        chatBody: ChatBody
+    ): StreamRequest[Either[OpenAIException, Stream[F, ChatChunkResponse]], Fs2Streams[F]] =
+      client
+        .createChatCompletion(Fs2Streams[F], chatBody)
+        .mapResponse(mapEventToResponse[F])
+  }
+
+  private def mapEventToResponse[F[_]: RaiseThrowable](
+      response: Either[OpenAIException, Stream[F, Byte]]
+  ): Either[OpenAIException, Stream[F, ChatChunkResponse]] =
+    response.map(
+      _.through(Fs2ServerSentEvents.parse)
+        .through(deserializeEvent)
+        .rethrow
+    )
+
+  private def deserializeEvent[F[_]]: Pipe[F, ServerSentEvent, Either[OpenAIException, ChatChunkResponse]] =
+    _.collect {
+      case ServerSentEvent(Some(data), _, _, _) if data != DoneEventMessage =>
+        deserializeJsonSnake[ChatChunkResponse].apply(data)
+    }
+}


Test patch:
diff --git a/core/src/test/scala/sttp/openai/client/SyncClientSpec.scala b/core/src/test/scala/sttp/openai/client/SyncClientSpec.scala
index a8cd831..621feb9 100644
--- a/core/src/test/scala/sttp/openai/client/SyncClientSpec.scala
+++ b/core/src/test/scala/sttp/openai/client/SyncClientSpec.scala
@@ -4,47 +4,23 @@ import org.scalatest.EitherValues
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import sttp.client4._
-import sttp.model.StatusCode
 import sttp.model.StatusCode._
 import sttp.openai.OpenAIExceptions.OpenAIException
-import sttp.openai.OpenAIExceptions.OpenAIException._
 import sttp.openai.OpenAISyncClient
+import sttp.openai.fixtures.ErrorFixture
 import sttp.openai.requests.models.ModelsResponseData._
 
 class SyncClientSpec extends AnyFlatSpec with Matchers with EitherValues {
-  private val (message, typ, param, code) = ("Some error message.", "error_type", "null", "invalid_api_key")
-  private val errorResponse =
-    s"""
-      |{
-      |  "error": {
-      |    "message": "$message",
-      |    "type": "$typ",
-      |    "param": $param,
-      |    "code": "$code"
-      |  }
-      |}""".stripMargin
-  private val testData: Seq[(StatusCode, OpenAIException)] = List(
-    (TooManyRequests, new RateLimitException(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, TooManyRequests))),
-    (BadRequest, new InvalidRequestException(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, BadRequest))),
-    (NotFound, new InvalidRequestException(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, NotFound))),
-    (
-      UnsupportedMediaType,
-      new InvalidRequestException(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, UnsupportedMediaType))
-    ),
-    (Unauthorized, new AuthenticationException(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, Unauthorized))),
-    (Forbidden, new PermissionException(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, Forbidden))),
-    (Conflict, new TryAgain(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, Conflict))),
-    (
-      ServiceUnavailable,
-      new ServiceUnavailableException(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, ServiceUnavailable))
-    ),
-    (Gone, new APIException(Some(message), Some(typ), None, Some(code), HttpError(errorResponse, Gone)))
-  )
-  for ((statusCode, expectedError) <- testData)
+  for ((statusCode, expectedError) <- ErrorFixture.testData)
     s"Service response with status code: $statusCode" should s"return properly deserialized ${expectedError.getClass.getSimpleName}" in {
-      val syncBackendStub = DefaultSyncBackend.stub.whenAnyRequest.thenRespondWithCode(statusCode, errorResponse)
+      // given
+      val syncBackendStub = DefaultSyncBackend.stub.whenAnyRequest.thenRespondWithCode(statusCode, ErrorFixture.errorResponse)
       val syncClient = OpenAISyncClient(authToken = "test-token", backend = syncBackendStub)
+
+      // when
       val caught = intercept[OpenAIException](syncClient.getModels)
+
+      // then
       caught.getClass shouldBe expectedError.getClass
       caught.message shouldBe expectedError.message
       caught.cause shouldBe expectedError.cause
@@ -54,6 +30,7 @@ class SyncClientSpec extends AnyFlatSpec with Matchers with EitherValues {
     }
 
   "Fetching models with successful response" should "return properly deserialized list of available models" in {
+    // given
     val modelsResponse = sttp.openai.fixtures.ModelsGetResponse.singleModelResponse
     val syncBackendStub = DefaultSyncBackend.stub.whenAnyRequest.thenRespondWithCode(Ok, modelsResponse)
     val syncClient = OpenAISyncClient(authToken = "test-token", backend = syncBackendStub)
@@ -86,6 +63,8 @@ class SyncClientSpec extends AnyFlatSpec with Matchers with EitherValues {
         )
       )
     )
+
+    // when & then
     syncClient.getModels shouldBe deserializedModels
   }
 }
diff --git a/core/src/test/scala/sttp/openai/fixtures/ChatChunkFixture.scala b/core/src/test/scala/sttp/openai/fixtures/ChatChunkFixture.scala
new file mode 100644
index 0000000..f6f8705
--- /dev/null
+++ b/core/src/test/scala/sttp/openai/fixtures/ChatChunkFixture.scala
@@ -0,0 +1,52 @@
+package sttp.openai.fixtures
+
+object ChatChunkFixture {
+
+  val jsonResponse: String =
+    """{
+      |  "id": "chatcmpl-76FxnKOjnPkDVYTAQ1wK8iUNFJPvR",
+      |  "object": "chat.completion",
+      |  "created": 1681725687,
+      |  "model": "gpt-3.5-turbo-0301",
+      |  "choices": [
+      |    {
+      |      "delta": {
+      |        "role": "assistant",
+      |        "content": "  Hi",
+      |        "function_call": {
+      |          "arguments": "args",
+      |          "name": "Fish"
+      |        }
+      |      },
+      |      "finish_reason": "stop",
+      |      "index": 0
+      |    }
+      |  ]
+      |}""".stripMargin
+
+  val jsonRequest: String =
+    """{
+      |  "model": "gpt-3.5-turbo",
+      |  "messages": [
+      |    {
+      |      "role": "user",
+      |      "content": "Hello!",
+      |      "name": "Andrzej",
+      |      "function_call": {
+      |        "arguments": "args",
+      |        "name": "Fish"
+      |      }
+      |    }
+      |  ],
+      |  "temperature": 1,
+      |  "top_p": 1,
+      |  "n": 1,
+      |  "stop": "\n",
+      |  "max_tokens": 7,
+      |  "presence_penalty": 0,
+      |  "frequency_penalty": 0,
+      |  "user": "testUser",
+      |  "stream": true
+      |}""".stripMargin
+
+}
diff --git a/core/src/test/scala/sttp/openai/fixtures/ChatFixture.scala b/core/src/test/scala/sttp/openai/fixtures/ChatFixture.scala
index df9e0e8..2c6e532 100644
--- a/core/src/test/scala/sttp/openai/fixtures/ChatFixture.scala
+++ b/core/src/test/scala/sttp/openai/fixtures/ChatFixture.scala
@@ -3,17 +3,27 @@ package sttp.openai.fixtures
 object ChatFixture {
 
   val jsonRequest: String = """{
-      |  "model": "gpt-3.5-turbo",
-      |  "messages": [{"role": "user", "content": "Hello!", "name": "Andrzej"}],
-      |  "temperature": 1,
-      |  "top_p": 1,
-      |  "n": 1,
-      |  "stop": "\n",
-      |  "max_tokens": 7,
-      |  "presence_penalty": 0,
-      |  "frequency_penalty": 0,
-      |  "user": "testUser"
-      |}""".stripMargin
+     |  "model": "gpt-3.5-turbo",
+     |  "messages": [
+     |    {
+     |      "role": "user",
+     |      "content": "Hello!",
+     |      "name": "Andrzej",
+     |      "function_call": {
+     |        "arguments": "args",
+     |        "name": "Fish"
+     |      }
+     |    }
+     |  ],
+     |  "temperature": 1,
+     |  "top_p": 1,
+     |  "n": 1,
+     |  "stop": "\n",
+     |  "max_tokens": 7,
+     |  "presence_penalty": 0,
+     |  "frequency_penalty": 0,
+     |  "user": "testUser"
+     |}""".stripMargin
 
   val jsonResponse: String = """{
       |   "id": "chatcmpl-76FxnKOjnPkDVYTAQ1wK8iUNFJPvR",
@@ -30,7 +40,10 @@ object ChatFixture {
       |       "message": {
       |         "role": "assistant",
       |         "content": "Hi there! How can I assist you today?",
-      |         "name": "Fish"
+      |         "function_call": {
+      |           "arguments": "args",
+      |           "name": "Fish"
+      |         }
       |       },
       |       "finish_reason": "stop",
       |       "index": 0
diff --git a/core/src/test/scala/sttp/openai/fixtures/ErrorFixture.scala b/core/src/test/scala/sttp/openai/fixtures/ErrorFixture.scala
new file mode 100644
index 0000000..298beab
--- /dev/null
+++ b/core/src/test/scala/sttp/openai/fixtures/ErrorFixture.scala
@@ -0,0 +1,61 @@
+package sttp.openai.fixtures
+
+import sttp.client4._
+import sttp.model.StatusCode
+import sttp.model.StatusCode._
+import sttp.openai.OpenAIExceptions.OpenAIException
+import sttp.openai.OpenAIExceptions.OpenAIException._
+
+object ErrorFixture {
+  private val (message, errorType, param, code) = ("Some error message.", "error_type", "null", "invalid_api_key")
+
+  val errorResponse =
+    s"""
+      |{
+      |  "error": {
+      |    "message": "$message",
+      |    "type": "$errorType",
+      |    "param": $param,
+      |    "code": "$code"
+      |  }
+      |}""".stripMargin
+
+  val testData: Seq[(StatusCode, OpenAIException)] = List(
+    (
+      TooManyRequests,
+      new RateLimitException(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, TooManyRequests))
+    ),
+    (
+      BadRequest,
+      new InvalidRequestException(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, BadRequest))
+    ),
+    (
+      NotFound,
+      new InvalidRequestException(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, NotFound))
+    ),
+    (
+      UnsupportedMediaType,
+      new InvalidRequestException(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, UnsupportedMediaType))
+    ),
+    (
+      Unauthorized,
+      new AuthenticationException(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, Unauthorized))
+    ),
+    (
+      Forbidden,
+      new PermissionException(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, Forbidden))
+    ),
+    (
+      Conflict,
+      new TryAgain(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, Conflict))
+    ),
+    (
+      ServiceUnavailable,
+      new ServiceUnavailableException(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, ServiceUnavailable))
+    ),
+    (
+      Gone,
+      new APIException(Some(message), Some(errorType), None, Some(code), HttpError(errorResponse, Gone))
+    )
+  )
+}
diff --git a/core/src/test/scala/sttp/openai/requests/completions/chat/ChatChunkDataSpec.scala b/core/src/test/scala/sttp/openai/requests/completions/chat/ChatChunkDataSpec.scala
new file mode 100644
index 0000000..c4aea24
--- /dev/null
+++ b/core/src/test/scala/sttp/openai/requests/completions/chat/ChatChunkDataSpec.scala
@@ -0,0 +1,92 @@
+package sttp.openai.requests.completions.chat
+
+import org.scalatest.EitherValues
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import sttp.openai.fixtures
+import sttp.openai.json.SttpUpickleApiExtension
+import sttp.openai.requests.completions.Stop.SingleStop
+
+class ChatChunkDataSpec extends AnyFlatSpec with Matchers with EitherValues {
+
+  "Given chat chunk completions response as Json" should "be properly deserialized to case class" in {
+    import ChatChunkRequestResponseData._
+    import ChatChunkRequestResponseData.ChatChunkResponse._
+
+    // given
+    val jsonResponse = fixtures.ChatChunkFixture.jsonResponse
+
+    val functionCall: FunctionCall = FunctionCall(
+      arguments = "args",
+      name = "Fish"
+    )
+
+    val delta: Delta = Delta(
+      role = Some(Role.Assistant),
+      content = Some("  Hi"),
+      functionCall = Some(functionCall)
+    )
+
+    val choices: Choices = Choices(
+      delta = delta,
+      finishReason = "stop",
+      index = 0
+    )
+
+    val expectedResponse: ChatChunkResponse = ChatChunkResponse(
+      id = "chatcmpl-76FxnKOjnPkDVYTAQ1wK8iUNFJPvR",
+      `object` = "chat.completion",
+      created = 1681725687,
+      model = "gpt-3.5-turbo-0301",
+      choices = Seq(choices)
+    )
+
+    // when
+    val givenResponse: Either[Exception, ChatChunkResponse] = SttpUpickleApiExtension.deserializeJsonSnake.apply(jsonResponse)
+
+    // then
+    givenResponse.value shouldBe expectedResponse
+  }
+
+  "Given completions request with streaming enabled as case class" should "be properly serialized to Json" in {
+    import ChatRequestBody._
+
+    // given
+    val functionCall: FunctionCall = FunctionCall(
+      arguments = "args",
+      name = "Fish"
+    )
+
+    val messages: Seq[Message] = Seq(
+      Message(
+        role = Role.User,
+        content = "Hello!",
+        name = Some("Andrzej"),
+        functionCall = Some(functionCall)
+      )
+    )
+
+    val givenRequest = ChatRequestBody.ChatBody(
+      model = ChatCompletionModel.GPT35Turbo,
+      messages = messages,
+      temperature = Some(1),
+      topP = Some(1),
+      n = Some(1),
+      stop = Some(SingleStop("\n")),
+      maxTokens = Some(7),
+      presencePenalty = Some(0),
+      frequencyPenalty = Some(0),
+      user = Some("testUser")
+    )
+
+    val jsonRequest: ujson.Value = ujson.read(fixtures.ChatChunkFixture.jsonRequest)
+
+    // when
+    val serializedJson: ujson.Value = ChatBody.withStreaming(givenRequest)
+
+    // then
+    serializedJson shouldBe jsonRequest
+
+  }
+
+}
diff --git a/core/src/test/scala/sttp/openai/requests/completions/chat/ChatDataSpec.scala b/core/src/test/scala/sttp/openai/requests/completions/chat/ChatDataSpec.scala
index a9ae27c..459081c 100644
--- a/core/src/test/scala/sttp/openai/requests/completions/chat/ChatDataSpec.scala
+++ b/core/src/test/scala/sttp/openai/requests/completions/chat/ChatDataSpec.scala
@@ -4,16 +4,16 @@ import org.scalatest.EitherValues
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import sttp.openai.fixtures
-import sttp.openai.json.{SnakePickle, SttpUpickleApiExtension}
-import sttp.openai.requests.completions.Usage
+import sttp.openai.json.SnakePickle
+import sttp.openai.json.SttpUpickleApiExtension
 import sttp.openai.requests.completions.Stop.SingleStop
-import sttp.openai.requests.completions.chat.ChatRequestBody.ChatCompletionModel
+import sttp.openai.requests.completions.Usage
 
 class ChatDataSpec extends AnyFlatSpec with Matchers with EitherValues {
 
   "Given chat completions response as Json" should "be properly deserialized to case class" in {
-    import sttp.openai.requests.completions.chat.ChatRequestResponseData._
-    import sttp.openai.requests.completions.chat.ChatRequestResponseData.ChatResponse._
+    import ChatRequestResponseData._
+    import ChatRequestResponseData.ChatResponse._
 
     // given
     val jsonResponse = fixtures.ChatFixture.jsonResponse
@@ -24,10 +24,15 @@ class ChatDataSpec extends AnyFlatSpec with Matchers with EitherValues {
       totalTokens = 20
     )
 
+    val functionCall: FunctionCall = FunctionCall(
+      arguments = "args",
+      name = "Fish"
+    )
+
     val message: Message = Message(
       role = Role.Assistant,
       content = "Hi there! How can I assist you today?",
-      name = Some("Fish")
+      functionCall = Some(functionCall)
     )
 
     val choices: Choices = Choices(
@@ -53,12 +58,20 @@ class ChatDataSpec extends AnyFlatSpec with Matchers with EitherValues {
   }
 
   "Given completions request as case class" should "be properly serialized to Json" in {
+    import ChatRequestBody._
+
     // given
+    val functionCall: FunctionCall = FunctionCall(
+      arguments = "args",
+      name = "Fish"
+    )
+
     val messages: Seq[Message] = Seq(
       Message(
         role = Role.User,
         content = "Hello!",
-        name = Some("Andrzej")
+        name = Some("Andrzej"),
+        functionCall = Some(functionCall)
       )
     )
 
diff --git a/streaming/fs2/src/test/scala/sttp/openai/streaming/fs2/Fs2ClientSpec.scala b/streaming/fs2/src/test/scala/sttp/openai/streaming/fs2/Fs2ClientSpec.scala
new file mode 100644
index 0000000..caeac79
--- /dev/null
+++ b/streaming/fs2/src/test/scala/sttp/openai/streaming/fs2/Fs2ClientSpec.scala
@@ -0,0 +1,111 @@
+package sttp.openai.streaming.fs2
+
+import org.scalatest.EitherValues
+import org.scalatest.flatspec.AsyncFlatSpec
+import cats.effect.testing.scalatest.AsyncIOSpec
+import org.scalatest.matchers.should.Matchers
+import cats.effect.IO
+import _root_.fs2.{text, Stream}
+import sttp.client4.DeserializationException
+import sttp.client4.httpclient.fs2.HttpClientFs2Backend
+import sttp.client4.testing.RawStream
+import sttp.model.sse.ServerSentEvent
+import sttp.openai.OpenAI
+import sttp.openai.fixtures.ErrorFixture
+import sttp.openai.json.SnakePickle._
+import sttp.openai.requests.completions.chat.ChatChunkRequestResponseData.ChatChunkResponse
+import sttp.openai.requests.completions.chat.ChatChunkRequestResponseData.ChatChunkResponse.DoneEventMessage
+import sttp.openai.requests.completions.chat.ChatRequestBody.{ChatBody, ChatCompletionModel}
+import sttp.openai.streaming.fs2._
+
+class Fs2ClientSpec extends AsyncFlatSpec with AsyncIOSpec with Matchers with EitherValues {
+  for ((statusCode, expectedError) <- ErrorFixture.testData)
+    s"Service response with status code: $statusCode" should s"return properly deserialized ${expectedError.getClass.getSimpleName}" in {
+      // given
+      val fs2BackendStub = HttpClientFs2Backend.stub[IO].whenAnyRequest.thenRespondWithCode(statusCode, ErrorFixture.errorResponse)
+      val client = new OpenAI("test-token")
+
+      val givenRequest = ChatBody(
+        model = ChatCompletionModel.GPT35Turbo,
+        messages = Seq.empty
+      )
+
+      // when
+      val caught = client
+        .createStreamedChatCompletion[IO](givenRequest)
+        .send(fs2BackendStub)
+        .map(_.body.left.value)
+
+      // then
+      caught.asserting { c =>
+        c.getClass shouldBe expectedError.getClass
+        c.message shouldBe expectedError.message
+        c.cause shouldBe expectedError.cause
+        c.code shouldBe expectedError.code
+        c.param shouldBe expectedError.param
+        c.`type` shouldBe expectedError.`type`
+      }
+    }
+
+  "Creating chat completions with failed stream due to invalid deserialization" should "return properly deserialized error" in {
+    // given
+    val invalidJson = Some("invalid json")
+    val invalidEvent = ServerSentEvent(invalidJson)
+
+    val streamedResponse = Stream
+      .emit(invalidEvent.toString)
+      .through(text.utf8.encode)
+      .covary[IO]
+
+    val fs2BackendStub = HttpClientFs2Backend.stub[IO].whenAnyRequest.thenRespond(RawStream(streamedResponse))
+    val client = new OpenAI(authToken = "test-token")
+
+    val givenRequest = ChatBody(
+      model = ChatCompletionModel.GPT35Turbo,
+      messages = Seq.empty
+    )
+
+    // when
+    val response = client
+      .createStreamedChatCompletion[IO](givenRequest)
+      .send(fs2BackendStub)
+      .map(_.body.value)
+      .flatMap(_.compile.drain)
+
+    // then
+    response.attempt.asserting(_ shouldBe a[Left[DeserializationException[_], _]])
+  }
+
+  "Creating chat completions with successful response" should "return properly deserialized list of chunks" in {
+    // given
+    val chatChunks = Seq.fill(3)(sttp.openai.fixtures.ChatChunkFixture.jsonResponse).map(compactJson)
+    val events = chatChunks.map(data => ServerSentEvent(Some(data))) :+ ServerSentEvent(Some(DoneEventMessage))
+    val delimiter = "\n\n"
+    val streamedResponse = Stream
+      .emits(events)
+      .map(_.toString + delimiter)
+      .through(text.utf8.encode)
+      .covary[IO]
+
+    val fs2BackendStub = HttpClientFs2Backend.stub[IO].whenAnyRequest.thenRespond(RawStream(streamedResponse))
+    val client = new OpenAI(authToken = "test-token")
+
+    val givenRequest = ChatBody(
+      model = ChatCompletionModel.GPT35Turbo,
+      messages = Seq.empty
+    )
+
+    // when
+    val response = client
+      .createStreamedChatCompletion[IO](givenRequest)
+      .send(fs2BackendStub)
+      .map(_.body.value)
+      .flatMap(_.compile.toList)
+
+    // then
+    val expectedResponse = chatChunks.map(read[ChatChunkResponse](_))
+    response.asserting(_ shouldBe expectedResponse)
+  }
+
+  private def compactJson(json: String): String = write(read[ujson.Value](json))
+}