feat(agents): run local tools during session creation

docs/helpers/agent-sessions.md

@@ -26,6 +26,30 @@ try (StreamResponse<AgentSessionEvent> stream = client.beta().agents().sessions(
 }
 ```
 
+For the initial prompt, pass the same local handlers to `createStreaming`:
+
+```java
+import com.openai.models.beta.agents.AgentToolHandlers;
+import com.openai.models.beta.agents.sessions.SessionCreateParams;
+import com.openai.services.beta.agents.AgentTurnResults;
+
+var handlers = AgentToolHandlers.builder()
+    .toolHandler("weather", args -> weatherService.lookup(args))
+    .build();
+var creation = SessionCreateParams.builder()
+    .agent(agent) // Configure the corresponding function definition on the agent.
+    .environmentNone()
+    .input("Check the weather in Paris")
+    .build();
+
+try (var stream = client.beta().agents().sessions().createStreaming(creation, handlers)) {
+    System.out.println(AgentTurnResults.getFinalResult(stream).outputText());
+}
+```
+
+The async service accepts the same overload; use `asyncToolHandler` for callbacks
+returning a `CompletionStage`. Handlers run locally while consuming the stream.
+
 A handler receives a deep copy of the JSON object arguments and can return a
 string, JSON object (`Map`), `List<InputContentParam>`,
 `AgentFunctionCallOutputParam`, or `null`. Invalid arguments and handler exceptions

openai-java-core/src/main/kotlin/com/openai/models/beta/agents/AgentToolHandlers.kt

@@ -0,0 +1,32 @@
+package com.openai.models.beta.agents
+
+import java.util.concurrent.CompletableFuture
+import java.util.concurrent.CompletionStage
+import java.util.function.Function
+
+/** Local function handlers for a beta Agents creation stream. Never sent to the API. */
+class AgentToolHandlers
+private constructor(
+    internal val handlers: Map<String, Function<Map<String, Any?>, CompletionStage<*>>>
+) {
+    companion object {
+        @JvmStatic fun builder() = Builder()
+    }
+
+    class Builder {
+        private val handlers =
+            linkedMapOf<String, Function<Map<String, Any?>, CompletionStage<*>>>()
+
+        /** Uses the same return values and sanitized errors as session stream tool handlers. */
+        fun toolHandler(name: String, handler: Function<Map<String, Any?>, Any?>) = apply {
+            handlers[name] = Function { CompletableFuture.completedFuture(handler.apply(it)) }
+        }
+
+        fun asyncToolHandler(
+            name: String,
+            handler: Function<Map<String, Any?>, CompletionStage<*>>,
+        ) = apply { handlers[name] = handler }
+
+        fun build() = AgentToolHandlers(handlers.toMap())
+    }
+}

openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsync.kt

@@ -12,6 +12,7 @@ import com.openai.models.beta.agents.AgentSession
 import com.openai.models.beta.agents.AgentSessionDeleted
 import com.openai.models.beta.agents.AgentSessionEvent
 import com.openai.models.beta.agents.AgentSessionStreamParams
+import com.openai.models.beta.agents.AgentToolHandlers
 import com.openai.models.beta.agents.sessions.SessionCreateParams
 import com.openai.models.beta.agents.sessions.SessionDeleteParams
 import com.openai.models.beta.agents.sessions.SessionListPageAsync
@@ -94,6 +95,23 @@ interface SessionServiceAsync {
         requestOptions: RequestOptions = RequestOptions.none(),
     ): AsyncStreamResponse<AgentSessionEvent>
 
+    /** Creates and handles the initial turn using local function callbacks. */
+    fun createStreaming(
+        params: SessionCreateParams,
+        handlers: AgentToolHandlers,
+    ): AsyncStreamResponse<AgentSessionEvent> =
+        createStreaming(params, handlers, RequestOptions.none())
+
+    /** @see createStreaming */
+    fun createStreaming(
+        params: SessionCreateParams,
+        handlers: AgentToolHandlers,
+        requestOptions: RequestOptions,
+    ): AsyncStreamResponse<AgentSessionEvent> =
+        throw UnsupportedOperationException(
+            "This service does not implement creation tool handlers"
+        )
+
     /**
      * Retrieves the current state of a managed agent session. See
      * [managing sessions](https://developers.openai.com/api/docs/guides/agents-api/sessions/manage).

openai-java-core/src/main/kotlin/com/openai/services/async/beta/agents/SessionServiceAsyncImpl.kt

@@ -29,6 +29,7 @@ import com.openai.models.beta.agents.AgentSession
 import com.openai.models.beta.agents.AgentSessionDeleted
 import com.openai.models.beta.agents.AgentSessionEvent
 import com.openai.models.beta.agents.AgentSessionStreamParams
+import com.openai.models.beta.agents.AgentToolHandlers
 import com.openai.models.beta.agents.sessions.SessionCreateParams
 import com.openai.models.beta.agents.sessions.SessionDeleteParams
 import com.openai.models.beta.agents.sessions.SessionListPageAsync
@@ -129,6 +130,24 @@ class SessionServiceAsyncImpl internal constructor(private val clientOptions: Cl
             .toAsync(clientOptions.streamHandlerExecutor)
             .let { AgentTurnResults.collecting(it) }
 
+    override fun createStreaming(
+        params: SessionCreateParams,
+        handlers: AgentToolHandlers,
+        requestOptions: RequestOptions,
+    ): AsyncStreamResponse<AgentSessionEvent> =
+        AgentTurnResults.collecting(
+            AgentSessionStreamAsync(
+                this,
+                null,
+                requestOptions,
+                clientOptions.streamHandlerExecutor,
+                clientOptions.sleeper,
+                params,
+                handlers,
+            ),
+            handlers.handlers.keys,
+        )
+
     override fun retrieve(
         params: SessionRetrieveParams,
         requestOptions: RequestOptions,

openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStream.kt

@@ -4,6 +4,8 @@ import com.openai.core.RequestOptions
 import com.openai.core.http.StreamResponse
 import com.openai.models.beta.agents.AgentSessionEvent
 import com.openai.models.beta.agents.AgentSessionStreamParams
+import com.openai.models.beta.agents.AgentToolHandlers
+import com.openai.models.beta.agents.sessions.SessionCreateParams
 import com.openai.models.beta.agents.sessions.events.EventCreateParams
 import com.openai.services.blocking.beta.agents.SessionService
 import java.util.Spliterator
@@ -15,17 +17,22 @@ import java.util.stream.Stream
 import java.util.stream.StreamSupport
 
 /**
- * One turn's original typed events. Requires an idle session and a single input writer. Use
- * try-with-resources. Closing releases the connection without cancelling the backend turn. Unknown
- * tools remain manual. Registered handlers run sequentially after their event is delivered.
+ * One turn's original typed events. Follow-up streams require an idle session and a single input
+ * writer. Use try-with-resources. Closing releases the connection without cancelling the backend
+ * turn. Unknown tools remain manual. Registered handlers run sequentially after their event is
+ * delivered.
  */
 internal class AgentSessionStream(
     private val sessions: SessionService,
-    params: AgentSessionStreamParams,
+    params: AgentSessionStreamParams?,
     private val options: RequestOptions,
     private val sleeper: com.openai.core.Sleeper,
+    private val creation: SessionCreateParams? = null,
+    handlers: AgentToolHandlers? = null,
 ) : StreamResponse<AgentSessionEvent> {
-    private val support = AgentSessionStreamSupport(params)
+    private val support =
+        if (creation == null) AgentSessionStreamSupport(checkNotNull(params))
+        else AgentSessionStreamSupport(checkNotNull(handlers), creation)
     private val closed = AtomicBoolean()
     private val submissionLock = Any()
     private val consumed = AtomicBoolean()
@@ -35,18 +42,22 @@ internal class AgentSessionStream(
     private val response: com.openai.core.http.HttpResponseFor<StreamResponse<AgentSessionEvent>>
 
     init {
-        support.checkIdle(sessions.retrieve(support.retrieveParams(), options))
+        if (creation == null)
+            support.checkIdle(sessions.retrieve(support.retrieveParams(), options))
         response =
-            sessions.events().withRawResponse().streamStreaming(support.streamParams(), options)
+            if (creation == null)
+                sessions.events().withRawResponse().streamStreaming(support.streamParams(), options)
+            else sessions.withRawResponse().createStreaming(creation, options)
         source =
             try {
-                response.parse()
+                AgentTurnResults.uncollected(response.parse())
             } catch (error: Throwable) {
                 response.close()
                 throw error
             }
         try {
-            sessions.events().withRawResponse().create(support.inputParams(), options).close()
+            if (creation == null)
+                sessions.events().withRawResponse().create(support.inputParams(), options).close()
         } catch (error: Throwable) {
             close()
             throw error

openai-java-core/src/main/kotlin/com/openai/services/beta/agents/AgentSessionStreamAsync.kt

@@ -5,6 +5,8 @@ import com.openai.core.http.AsyncStreamResponse
 import com.openai.core.http.StreamResponse
 import com.openai.models.beta.agents.AgentSessionEvent
 import com.openai.models.beta.agents.AgentSessionStreamParams
+import com.openai.models.beta.agents.AgentToolHandlers
+import com.openai.models.beta.agents.sessions.SessionCreateParams
 import com.openai.models.beta.agents.sessions.events.EventCreateParams
 import com.openai.services.async.beta.agents.SessionServiceAsync
 import java.util.Optional
@@ -15,12 +17,16 @@ import java.util.concurrent.atomic.AtomicBoolean
 /** Async one-turn stream; uses the SDK executor and never allocates a thread pool. */
 internal class AgentSessionStreamAsync(
     private val sessions: SessionServiceAsync,
-    params: AgentSessionStreamParams,
+    params: AgentSessionStreamParams?,
     private val options: RequestOptions,
     private val defaultExecutor: Executor,
     private val sleeper: com.openai.core.Sleeper,
+    private val creation: SessionCreateParams? = null,
+    handlers: AgentToolHandlers? = null,
 ) : AsyncStreamResponse<AgentSessionEvent> {
-    private val support = AgentSessionStreamSupport(params)
+    private val support =
+        if (creation == null) AgentSessionStreamSupport(checkNotNull(params))
+        else AgentSessionStreamSupport(checkNotNull(handlers), creation)
     private val subscribed = AtomicBoolean()
     private val closed = AtomicBoolean()
     private val submissionLock = Any()
@@ -58,33 +64,48 @@ internal class AgentSessionStreamAsync(
             execute { notifyComplete(null) }
             return@apply
         }
-        await(sessions.retrieve(support.retrieveParams(), options), cancelOnClose = false) { session
-            ->
-            support.checkIdle(session)
-            val opening =
-                sessions.events().withRawResponse().streamStreaming(support.streamParams(), options)
-            // Closing during an in-flight open must also close the eventual HTTP response.
-            opening.whenComplete { response, _ ->
-                openingResponse = response
-                if (closed.get()) response?.close()
+        if (creation != null) open(sessions.withRawResponse().createStreaming(creation, options))
+        else
+            await(sessions.retrieve(support.retrieveParams(), options), cancelOnClose = false) {
+                session ->
+                support.checkIdle(session)
+                open(
+                    sessions
+                        .events()
+                        .withRawResponse()
+                        .streamStreaming(support.streamParams(), options)
+                )
             }
-            await(opening, cancelOnClose = false) { response ->
-                openingResponse = response
-                val stream =
-                    try {
-                        response.parse()
-                    } catch (error: Throwable) {
-                        response.close()
-                        throw error
-                    }
-                source = stream
-                if (closed.get()) {
-                    stream.close()
-                    return@await
+    }
+
+    private fun open(
+        opening:
+            CompletableFuture<
+                com.openai.core.http.HttpResponseFor<StreamResponse<AgentSessionEvent>>
+            >
+    ) {
+        // A response arriving after close must still be released.
+        opening.whenComplete { response, _ ->
+            openingResponse = response
+            if (closed.get()) response?.close()
+        }
+        await(opening, cancelOnClose = false) { response ->
+            openingResponse = response
+            val stream =
+                try {
+                    AgentTurnResults.uncollected(response.parse())
+                } catch (error: Throwable) {
+                    response.close()
+                    throw error
                 }
-                iterator = stream.stream().iterator()
-                create(support.inputParams())?.let { await(it) { pump() } }
+            source = stream
+            if (closed.get()) {
+                stream.close()
+                return@await
             }
+            iterator = stream.stream().iterator()
+            if (creation != null) pump()
+            else create(support.inputParams())?.let { await(it) { pump() } }
         }
     }