Skip to content

Support tool calling while streaming Open Responses #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ActuallyTaylor merged 2 commits into from
Aug 3, 2026
Merged

Support tool calling while streaming Open Responses #8

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
96b0bac
Support tool calling while streaming Open Responses
claude Aug 3, 2026
7b1327c
Let the Responses test suite target any compatible server
claude Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
180 changes: 159 additions & 21 deletions Sources/AnyLanguageModel/Models/OpenResponsesLanguageModel.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -420,17 +420,21 @@ public struct OpenResponsesLanguageModel: LanguageModel {
let url = baseURL.appendingPathComponent("responses")
let stream: AsyncThrowingStream<LanguageModelSession.ResponseStream<Content>.Snapshot, any Error> = .init {
continuation in
do {
let params = try OpenResponsesAPI.createRequestBody(
model: model,
messages: session.transcript.toOpenResponsesMessages(),
tools: tools,
generating: type,
options: options,
stream: true
)
let task = Task { @Sendable in
do {
let task = Task { @Sendable in
do {
var messages = session.transcript.toOpenResponsesMessages()

// Each iteration is one turn: stream a response, then execute any tools it requested
// and feed the results back for another turn.
turnLoop: while true {
let params = try OpenResponsesAPI.createRequestBody(
model: model,
messages: messages,
tools: tools,
generating: type,
options: options,
stream: true
)
let body = try JSONEncoder().encode(params)
let events: AsyncThrowingStream<OpenResponsesStreamEvent, any Error> =
httpSession.fetchEventStream(
Expand All @@ -440,10 +444,18 @@ public struct OpenResponsesLanguageModel: LanguageModel {
body: body
)
var accumulatedText = ""
for try await event in events {
// Streamed output items, keyed by their output index. Function-call arguments
// arrive as fragments and are accumulated until the turn ends.
var streamedItems: [Int: OpenResponsesStreamingOutputItem] = [:]

eventStream: for try await event in events {
switch event {
case .outputTextDelta(let delta):
accumulatedText += delta

// Grow the observable transcript so a Transcript-driven UI updates live.
session.growStreamingTranscript(text: accumulatedText)

var raw: GeneratedContent
let content: Content.PartiallyGenerated?
if type == String.self {
Expand All @@ -458,25 +470,77 @@ public struct OpenResponsesLanguageModel: LanguageModel {
if let content {
continuation.yield(.init(content: content, rawContent: raw))
}
case .outputItemAdded(let index, let item):
// Opening event: carries the call id and name, with empty arguments.
streamedItems[index] = OpenResponsesStreamingOutputItem(item: item)
case .functionCallArgumentsDelta(let index, let delta):
// Only accumulate for items the server already opened.
guard streamedItems[index] != nil else { continue }
streamedItems[index]?.argumentFragments += delta
case .outputItemDone(let index, let item):
// Keep any fragments accumulated for this index; the finished item
// may or may not repeat the full arguments string.
let fragments = streamedItems[index]?.argumentFragments ?? ""
streamedItems[index] = OpenResponsesStreamingOutputItem(
item: item,
argumentFragments: fragments
)
case .completed:
continuation.finish()
return
// Need a label, otherwise this would only break out of the switch.
break eventStream
case .failed:
continuation.finish(throwing: OpenResponsesLanguageModelError.streamFailed)
return
case .ignored:
break
}
}
continuation.finish()
} catch {
continuation.finish(throwing: error)

let assembled = assembleOpenResponsesStreamedOutput(streamedItems)
// The turn is complete once the model stops asking for tools.
guard !assembled.toolCalls.isEmpty else { break turnLoop }

// Echo this turn's output items back so the follow-up request keeps the calls in context.
for item in assembled.items {
messages.append(OpenResponsesMessage(role: .raw(rawContent: item), content: .text("")))
}

let resolution = try await resolveOpenResponsesToolCalls(assembled.toolCalls, session: session)
switch resolution {
case .stop(let calls):
if !calls.isEmpty {
session.appendTranscriptEntry(.toolCalls(Transcript.ToolCalls(calls)))
}
continuation.finish()
return
case .invocations(let invocations):
guard !invocations.isEmpty else { break turnLoop }

// Tool calls must be appended before the outputs of the same turn.
session.appendTranscriptEntry(
.toolCalls(Transcript.ToolCalls(invocations.map { $0.call }))
)

for inv in invocations {
session.appendTranscriptEntry(.toolOutput(inv.output))
messages.append(
OpenResponsesMessage(
role: .tool(id: inv.call.id),
content: .text(
openResponsesConvertSegmentsToToolContentString(inv.output.segments)
)
)
)
}
}
}

continuation.finish()
} catch {
continuation.finish(throwing: error)
}
continuation.onTermination = { _ in task.cancel() }
} catch {
continuation.finish(throwing: error)
}
continuation.onTermination = { _ in task.cancel() }
}
return LanguageModelSession.ResponseStream(stream: stream)
}
Expand Down Expand Up @@ -1034,6 +1098,52 @@ private func extractJSONFromOutput(_ output: [JSONValue]?) -> String? {
return nil
}

/// An output item being assembled from a stream, along with any function-call argument
/// fragments that arrived for it.
private struct OpenResponsesStreamingOutputItem: Sendable {
var item: JSONValue
var argumentFragments: String = ""
}

/// Merges streamed argument fragments into their output items and pulls out the function calls of the turn.
///
/// Items are returned in output-index order so they can be echoed back as the assistant's turn.
/// A function call is only surfaced once its accumulated fragments form parseable JSON.
private func assembleOpenResponsesStreamedOutput(
_ streamedItems: [Int: OpenResponsesStreamingOutputItem]
) -> (items: [JSONValue], toolCalls: [OpenResponsesToolCall]) {
var items: [JSONValue] = []
var toolCalls: [OpenResponsesToolCall] = []

for entry in streamedItems.sorted(by: { $0.key < $1.key }).map(\.value) {
guard case .object(var obj) = entry.item else {
items.append(entry.item)
continue
}
let type = obj["type"].flatMap { if case .string(let s) = $0 { return s } else { return nil } }
guard type == "function_call" else {
items.append(entry.item)
continue
}

// Prefer the fragments we accumulated; fall back to whatever the finished item carried.
let fragments = entry.argumentFragments.trimmingCharacters(in: .whitespacesAndNewlines)
if !fragments.isEmpty {
obj["arguments"] = .string(fragments)
}
if case .string(let arguments)? = obj["arguments"], !arguments.isEmpty {
let data = arguments.data(using: .utf8)
guard data.flatMap({ try? JSONDecoder().decode(JSONValue.self, from: $0) }) != nil else { continue }
}

guard let call = parseOpenResponsesToolCall(from: obj) else { continue }
items.append(.object(obj))
toolCalls.append(call)
}

return (items, toolCalls)
}

private func resolveOpenResponsesToolCalls(
_ toolCalls: [OpenResponsesToolCall],
session: LanguageModelSession
Expand All @@ -1053,6 +1163,12 @@ private func resolveOpenResponsesToolCalls(

private enum OpenResponsesStreamEvent: Decodable, Sendable {
case outputTextDelta(String)
/// An output item was opened. For function calls this carries the call id and name.
case outputItemAdded(index: Int, item: JSONValue)
/// A fragment of a function call's arguments JSON.
case functionCallArgumentsDelta(index: Int, delta: String)
/// An output item finished, carrying its final form.
case outputItemDone(index: Int, item: JSONValue)
case completed
case failed
case ignored
Expand All @@ -1063,6 +1179,23 @@ private enum OpenResponsesStreamEvent: Decodable, Sendable {
switch type {
case "response.output_text.delta":
self = .outputTextDelta(try c.decode(String.self, forKey: .delta))
case "response.output_item.added":
guard let item = try c.decodeIfPresent(JSONValue.self, forKey: .item) else {
self = .ignored
return
}
self = .outputItemAdded(index: try c.decodeIfPresent(Int.self, forKey: .outputIndex) ?? 0, item: item)
case "response.function_call_arguments.delta":
self = .functionCallArgumentsDelta(
index: try c.decodeIfPresent(Int.self, forKey: .outputIndex) ?? 0,
delta: try c.decode(String.self, forKey: .delta)
)
case "response.output_item.done":
guard let item = try c.decodeIfPresent(JSONValue.self, forKey: .item) else {
self = .ignored
return
}
self = .outputItemDone(index: try c.decodeIfPresent(Int.self, forKey: .outputIndex) ?? 0, item: item)
case "response.completed":
self = .completed
case "response.failed":
Expand All @@ -1071,7 +1204,12 @@ private enum OpenResponsesStreamEvent: Decodable, Sendable {
self = .ignored
}
}
private enum CodingKeys: String, CodingKey { case type, delta }
private enum CodingKeys: String, CodingKey {
case type
case delta
case item
case outputIndex = "output_index"
}
}

// MARK: - Errors
Expand Down
52 changes: 50 additions & 2 deletions Tests/AnyLanguageModelTests/OpenResponsesLanguageModelTests.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,14 @@ private let openResponsesBaseURL: URL? = ProcessInfo.processInfo.environment["OP
URL(string: $0)
}

/// The model to exercise the Responses API with.
///
/// The endpoint is configurable, so the model has to be too — any server implementing the
/// Responses API serves its own model names. Ollama, for example, runs this suite with
/// `OPEN_RESPONSES_BASE_URL=http://localhost:11434/v1` and one of its local models.
private let openResponsesModel: String =
ProcessInfo.processInfo.environment["OPEN_RESPONSES_MODEL"] ?? "gpt-4o-mini"

@Suite("OpenResponsesLanguageModel")
struct OpenResponsesLanguageModelTests {
@Test func customHost() throws {
Expand All @@ -34,7 +42,7 @@ struct OpenResponsesLanguageModelTests {
OpenResponsesLanguageModel(
baseURL: baseURL,
apiKey: apiKey,
model: "gpt-4o-mini"
model: openResponsesModel
)
}

Expand Down Expand Up @@ -141,6 +149,46 @@ struct OpenResponsesLanguageModelTests {
#expect(foundToolOutput)
}

@Test func streamWithTools() async throws {
let weatherTool = WeatherTool()
let session = LanguageModelSession(model: model, tools: [weatherTool])

let stream = session.streamResponse(to: "How's the weather in San Francisco?")

var snapshots: [LanguageModelSession.ResponseStream<String>.Snapshot] = []

var toolAppearedInTranscript = false
var toolResponseAppearedInTranscript = false

for try await snapshot in stream {
snapshots.append(snapshot)

for entry in session.transcript {
switch entry {
case .toolCalls:
toolAppearedInTranscript = true
case .toolOutput:
toolResponseAppearedInTranscript = true
default: break
}
}
}

#expect(toolAppearedInTranscript, "Expected a tool call to appear in the transcript during streaming.")
#expect(
toolResponseAppearedInTranscript,
"Expected a tool output to appear in the transcript during streaming."
)

var foundToolOutput = false
for case let .toolOutput(toolOutput) in session.transcript {
#expect(!toolOutput.id.isEmpty)
#expect(toolOutput.toolName == "getWeather")
foundToolOutput = true
}
#expect(foundToolOutput, "Expected the 'getWeather' tool to exist in the final transcript.")
}

@Suite("Structured Output")
struct StructuredOutputTests {
@Generable
Expand Down Expand Up @@ -171,7 +219,7 @@ struct OpenResponsesLanguageModelTests {
OpenResponsesLanguageModel(
baseURL: openResponsesBaseURL!,
apiKey: openResponsesAPIKey!,
model: "gpt-4o-mini"
model: openResponsesModel
)
}

Expand Down