From edea87220add3f4731551d8b0ceddc1098f191a8 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Fri, 28 Aug 2026 00:51:19 +0200 Subject: [PATCH 01/15] feat: add MCP Tasks extension protocol types and client methods Add core protocol types for the experimental MCP Tasks extension (SEP-2663, `io.modelcontextprotocol/tasks`): TaskStatus, CreateTaskResult, GetTaskResult, the tasks/get, tasks/cancel and tasks/update request/param types, TaskAck and the notifications/tasks notification, with circe codecs matching the reference wire format. Add the requestor-side client methods getTask, cancelTask and updateTask to McpClient, so a client can poll a durable task handle returned by a task-supporting server, retrieve its result, cancel it, or answer an input request. Server-side task execution (returning a task from tools/call, task storage and driving) is intentionally left for a follow-up; this establishes the protocol vocabulary and requestor side first. Tracks #163. Co-Authored-By: Claude Opus 4.8 --- .../main/scala/chimp/client/McpClient.scala | 12 +++ .../scala/chimp/client/McpClientImpl.scala | 9 ++ .../scala/chimp/client/TasksClientSpec.scala | 65 ++++++++++++++ .../src/main/scala/chimp/protocol/Tasks.scala | 86 +++++++++++++++++++ .../test/scala/chimp/protocol/TasksSpec.scala | 50 +++++++++++ docs/client/capabilities.md | 16 ++++ 6 files changed, 238 insertions(+) create mode 100644 client/src/test/scala/chimp/client/TasksClientSpec.scala create mode 100644 core/src/main/scala/chimp/protocol/Tasks.scala create mode 100644 core/src/test/scala/chimp/protocol/TasksSpec.scala diff --git a/client/src/main/scala/chimp/client/McpClient.scala b/client/src/main/scala/chimp/client/McpClient.scala index 8c33c40..989e092 100644 --- a/client/src/main/scala/chimp/client/McpClient.scala +++ b/client/src/main/scala/chimp/client/McpClient.scala @@ -94,6 +94,18 @@ trait McpClient[F[_]]: */ def sendProgress(token: ProgressToken, progress: Double, total: Option[Double] = None, message: Option[String] = None): F[Unit] + /** Retrieves the current state of a task by its id (MCP Tasks extension, experimental). When a receiver answers `tools/call` with a + * [[chimp.protocol.CreateTaskResult]], the returned `taskId` is polled with this method until the task reaches a terminal state; the + * underlying result is then available in [[chimp.protocol.GetTaskResult.result]]. + */ + def getTask(taskId: String): F[GetTaskResult] + + /** Requests cancellation of a task by its id (MCP Tasks extension, experimental). */ + def cancelTask(taskId: String): F[Unit] + + /** Fulfils the input a task is waiting for while it is `InputRequired` (MCP Tasks extension, experimental). */ + def updateTask(taskId: String, inputResponses: Json): F[Unit] + /** An [[McpClient]] used over a [[chimp.client.transport.ClientBidirectionalTransport]], which additionally supports server-initiated * interactions: subscribing to resource updates, notifying the server about changes to the client's roots, and handling notifications * pushed by the server. diff --git a/client/src/main/scala/chimp/client/McpClientImpl.scala b/client/src/main/scala/chimp/client/McpClientImpl.scala index 0b96663..9c79d49 100644 --- a/client/src/main/scala/chimp/client/McpClientImpl.scala +++ b/client/src/main/scala/chimp/client/McpClientImpl.scala @@ -220,6 +220,15 @@ object McpClientImpl: val params = ProgressParams(progressToken = token, progress = progress, total = total, message = message).asJson sendNotification("notifications/progress", Some(params)) + override def getTask(taskId: String): F[GetTaskResult] = + sendRequest[GetTaskResult]("tasks/get", Some(GetTaskParams(taskId).asJson)) + + override def cancelTask(taskId: String): F[Unit] = + sendRequest[Json]("tasks/cancel", Some(CancelTaskParams(taskId).asJson)).map(_ => ()) + + override def updateTask(taskId: String, inputResponses: Json): F[Unit] = + sendRequest[Json]("tasks/update", Some(UpdateTaskParams(taskId, inputResponses).asJson)).map(_ => ()) + protected def requireServerCapability[A](method: String, present: ServerCapabilities => Boolean)(action: => F[A]): F[A] = if present(serverCapabilities) then action else monad.error(McpProtocolException(s"Server did not negotiate the capability required for $method")) diff --git a/client/src/test/scala/chimp/client/TasksClientSpec.scala b/client/src/test/scala/chimp/client/TasksClientSpec.scala new file mode 100644 index 0000000..bc33063 --- /dev/null +++ b/client/src/test/scala/chimp/client/TasksClientSpec.scala @@ -0,0 +1,65 @@ +package chimp.client + +import chimp.client.transport.ClientHttpTransport +import chimp.protocol.* +import io.circe.syntax.* +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import sttp.client4.testing.SyncBackendStub +import sttp.client4.{GenericRequest, StringBody} +import sttp.model.StatusCode +import sttp.shared.Identity + +class TasksClientSpec extends AnyFlatSpec with Matchers: + + private val mcpUri = sttp.model.Uri.parse("http://localhost/mcp").toOption.get + private val clientInfo = Implementation(name = "chimp-test", version = "0.0.1") + + private def envelopeFor(method: String, request: GenericRequest[?, ?]): Boolean = + request.body match + case StringBody(s, _, _) => s.contains(s"\"$method\"") + case _ => false + + private val initEnvelope: String = + val initResult = InitializeResult( + protocolVersion = ProtocolVersion.Latest.name, + capabilities = ServerCapabilities(), + serverInfo = Implementation(name = "test-server", version = "1.0") + ) + (JSONRPCMessage.Response(id = RequestId(1), result = initResult.asJson): JSONRPCMessage).asJson.noSpaces + + private def client(backend: sttp.client4.testing.SyncBackendStub): McpClient[Identity] = + McpClient[Identity](ClientHttpTransport[Identity](backend, mcpUri), clientInfo, ProtocolVersion.Latest) + + it should "poll a task with tasks/get and expose the underlying result" in: + val task = GetTaskResult( + taskId = "t1", + status = TaskStatus.Completed, + result = Some(CallToolResult(content = List(ToolContent.Text(text = "done"))).asJson), + resultType = Some("complete") + ) + val taskEnvelope = (JSONRPCMessage.Response(id = RequestId(1), result = task.asJson): JSONRPCMessage).asJson.noSpaces + val backend = SyncBackendStub + .whenRequestMatches(envelopeFor("initialize", _)) + .thenRespondAdjust(initEnvelope) + .whenRequestMatches(envelopeFor("tasks/get", _)) + .thenRespondAdjust(taskEnvelope) + .whenAnyRequest + .thenRespondAdjust("", StatusCode.Accepted) + + val res = client(backend).getTask("t1") + res.status shouldBe TaskStatus.Completed + res.result.flatMap(_.as[CallToolResult].toOption).map(_.content.head) shouldBe Some(ToolContent.Text("text", "done")) + + it should "cancel a task with tasks/cancel" in: + val ack = TaskAck(taskId = Some("t1"), status = Some(TaskStatus.Cancelled)) + val ackEnvelope = (JSONRPCMessage.Response(id = RequestId(1), result = ack.asJson): JSONRPCMessage).asJson.noSpaces + val backend = SyncBackendStub + .whenRequestMatches(envelopeFor("initialize", _)) + .thenRespondAdjust(initEnvelope) + .whenRequestMatches(envelopeFor("tasks/cancel", _)) + .thenRespondAdjust(ackEnvelope) + .whenAnyRequest + .thenRespondAdjust("", StatusCode.Accepted) + + noException should be thrownBy client(backend).cancelTask("t1") diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala new file mode 100644 index 0000000..21d5431 --- /dev/null +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -0,0 +1,86 @@ +package chimp.protocol + +import io.circe.{Codec, Decoder, Encoder, Json} + +/** The MCP Tasks extension (SEP-2663, identifier `io.modelcontextprotocol/tasks`): durable handles that let a receiver answer a request + * with a task, which the requestor then polls and later collects the result of. Experimental; the wire format follows the reference + * extension and may change. + */ +object TasksExtension: + val Id: String = "io.modelcontextprotocol/tasks" + +/** State of a task. Terminal states are `Completed`, `Failed` and `Cancelled`. */ +enum TaskStatus: + case Working, InputRequired, Completed, Failed, Cancelled + +object TaskStatus: + private val toWire: Map[TaskStatus, String] = Map( + Working -> "working", + InputRequired -> "input_required", + Completed -> "completed", + Failed -> "failed", + Cancelled -> "cancelled" + ) + private val fromWire: Map[String, TaskStatus] = toWire.map((k, v) => v -> k) + + def isTerminal(status: TaskStatus): Boolean = status match + case Completed | Failed | Cancelled => true + case Working | InputRequired => false + + given Encoder[TaskStatus] = Encoder.instance(status => Json.fromString(toWire(status))) + given Decoder[TaskStatus] = Decoder.decodeString.emap(s => fromWire.get(s).toRight(s"Unknown task status: $s")) + +/** Result returned when a receiver answers a request with a task instead of the request's normal result. */ +final case class CreateTaskResult( + taskId: String, + status: TaskStatus, + createdAt: Option[String] = None, + lastUpdatedAt: Option[String] = None, + ttlMs: Option[Long] = None, + pollIntervalMs: Option[Long] = None, + statusMessage: Option[String] = None, + resultType: String = "task", + _meta: Option[Map[String, Json]] = None +) derives Codec + +final case class GetTaskParams(taskId: String, _meta: Option[Map[String, Json]] = None) derives Codec +final case class GetTaskRequest(method: String = "tasks/get", params: GetTaskParams) derives Codec + +/** Detailed task state returned by `tasks/get`. `result` is present once the task is `Completed`, `error` once it has `Failed`, and + * `inputRequests` while it is `InputRequired`. `result` and `inputRequests` are left as raw JSON, since their shape depends on the request + * the task stands for. + */ +final case class GetTaskResult( + taskId: String, + status: TaskStatus, + createdAt: Option[String] = None, + lastUpdatedAt: Option[String] = None, + ttlMs: Option[Long] = None, + pollIntervalMs: Option[Long] = None, + statusMessage: Option[String] = None, + result: Option[Json] = None, + error: Option[Json] = None, + inputRequests: Option[Json] = None, + resultType: Option[String] = None, + _meta: Option[Map[String, Json]] = None +) derives Codec + +final case class CancelTaskParams(taskId: String, _meta: Option[Map[String, Json]] = None) derives Codec +final case class CancelTaskRequest(method: String = "tasks/cancel", params: CancelTaskParams) derives Codec + +final case class UpdateTaskParams(taskId: String, inputResponses: Json, _meta: Option[Map[String, Json]] = None) derives Codec +final case class UpdateTaskRequest(method: String = "tasks/update", params: UpdateTaskParams) derives Codec + +/** Acknowledgement returned by `tasks/cancel` and `tasks/update`. */ +final case class TaskAck( + taskId: Option[String] = None, + status: Option[TaskStatus] = None, + resultType: String = "complete", + _meta: Option[Map[String, Json]] = None +) derives Codec + +/** Notification pushed by a receiver that supports task subscriptions; carries the same fields as a `tasks/get` result. */ +final case class TaskStatusNotification( + method: String = "notifications/tasks", + params: GetTaskResult +) derives Codec diff --git a/core/src/test/scala/chimp/protocol/TasksSpec.scala b/core/src/test/scala/chimp/protocol/TasksSpec.scala new file mode 100644 index 0000000..2049501 --- /dev/null +++ b/core/src/test/scala/chimp/protocol/TasksSpec.scala @@ -0,0 +1,50 @@ +package chimp.protocol + +import io.circe.Json +import io.circe.parser.decode +import io.circe.syntax.* +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +class TasksSpec extends AnyFlatSpec with Matchers: + + it should "decode a CreateTaskResult from the spec example" in: + val json = + """ + { + "resultType": "task", + "taskId": "786512e2-9e0d-44bd-8f29-789f320fe840", + "status": "working", + "createdAt": "2025-11-25T10:30:00Z", + "lastUpdatedAt": "2025-11-25T10:50:00Z", + "ttlMs": 3600000, + "pollIntervalMs": 5000 + } + """ + val res = decode[CreateTaskResult](json) + res.map(_.taskId) shouldBe Right("786512e2-9e0d-44bd-8f29-789f320fe840") + res.map(_.status) shouldBe Right(TaskStatus.Working) + res.map(_.ttlMs) shouldBe Right(Some(3600000L)) + res.map(_.pollIntervalMs) shouldBe Right(Some(5000L)) + + it should "encode and decode task status with the spec wire strings" in: + (TaskStatus.InputRequired: TaskStatus).asJson shouldBe Json.fromString("input_required") + (TaskStatus.Cancelled: TaskStatus).asJson shouldBe Json.fromString("cancelled") + decode[TaskStatus](""" "completed" """.trim) shouldBe Right(TaskStatus.Completed) + + it should "reject an unknown task status" in: + decode[TaskStatus](""" "bogus" """.trim).isLeft shouldBe true + + it should "round-trip a completed GetTaskResult carrying the tool result" in: + val toolResult = CallToolResult(content = List(ToolContent.Text(text = "Hello, Luca!"))).asJson + val task = GetTaskResult( + taskId = "t1", + status = TaskStatus.Completed, + result = Some(toolResult), + resultType = Some("complete") + ) + decode[GetTaskResult](task.asJson.noSpaces) shouldBe Right(task) + + it should "mark only completed, failed and cancelled as terminal" in: + TaskStatus.values.filter(TaskStatus.isTerminal).toSet shouldBe + Set(TaskStatus.Completed, TaskStatus.Failed, TaskStatus.Cancelled) diff --git a/docs/client/capabilities.md b/docs/client/capabilities.md index bede01f..27c1a5a 100644 --- a/docs/client/capabilities.md +++ b/docs/client/capabilities.md @@ -43,3 +43,19 @@ def listen(client: BidirectionalMcpClient[Task]): Task[Unit] = case _ => ZIO.unit } ``` + +## Tasks (experimental) + +The [Tasks extension](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/main/seps/2663-tasks-extension.md) (`io.modelcontextprotocol/tasks`) lets a server answer a long-running request with a durable task handle that the client polls and later collects the result of. The client supports the requestor side with `getTask`, `cancelTask` and `updateTask`. This part of the protocol is experimental and its wire format may change. + +```scala mdoc:compile-only +import chimp.client.* +import chimp.protocol.* +import zio.{Task, ZIO} + +def awaitResult(client: McpClient[Task], taskId: String): Task[GetTaskResult] = + client.getTask(taskId).flatMap { task => + if TaskStatus.isTerminal(task.status) then ZIO.succeed(task) + else awaitResult(client, taskId) + } +``` From 11e18812b3e4fc589386d90340808a9b5778f435 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Fri, 28 Aug 2026 09:45:54 +0200 Subject: [PATCH 02/15] feat: implement server-side task execution for the Tasks extension Complete the Tasks extension (io.modelcontextprotocol/tasks) with the server-side subsystem and client augmentation on top of the protocol types: Server: - TaskStore[F] (in-memory default) and TaskExecutor[F] (thread-pool executor for synchronous servers) with best-effort cancellation. - TaskSupport bundle attached via McpServer/StreamingMcpServer.withTasks, with per-tool useTask / requireTask policies. - McpHandler answers tools/call with a CreateTaskResult when the client declares the tasks capability in _meta and runs the tool in the background, transitioning the task to completed (with the CallToolResult) or failed; a late completion never overwrites a cancellation. Handles tasks/get, tasks/cancel and tasks/update, returns -32003 when a required task capability is absent, and advertises the extension in capabilities. Client: - callToolWithTasks declares the capability and returns either a CallToolResult or a CreateTaskResult task handle. Core: - MissingRequiredClientCapability (-32003), ServerCapabilities.extensions, and the client-capability _meta helpers. Server-initiated input_required flows remain a follow-up. Tracks #163. Co-Authored-By: Claude Opus 4.8 --- .../main/scala/chimp/client/McpClient.scala | 6 + .../scala/chimp/client/McpClientImpl.scala | 14 ++ .../scala/chimp/client/TasksClientSpec.scala | 19 +++ .../main/scala/chimp/protocol/JsonRpc.scala | 1 + .../main/scala/chimp/protocol/Lifecycle.scala | 3 +- .../src/main/scala/chimp/protocol/Tasks.scala | 14 ++ docs/server/capabilities.md | 25 ++++ .../main/scala/chimp/server/McpHandler.scala | 137 ++++++++++++++++-- .../main/scala/chimp/server/McpServer.scala | 16 +- .../main/scala/chimp/server/TaskSupport.scala | 74 ++++++++++ .../scala/chimp/server/TaskServerSpec.scala | 127 ++++++++++++++++ 11 files changed, 419 insertions(+), 17 deletions(-) create mode 100644 server/src/main/scala/chimp/server/TaskSupport.scala create mode 100644 server/src/test/scala/chimp/server/TaskServerSpec.scala diff --git a/client/src/main/scala/chimp/client/McpClient.scala b/client/src/main/scala/chimp/client/McpClient.scala index 989e092..ceef2f1 100644 --- a/client/src/main/scala/chimp/client/McpClient.scala +++ b/client/src/main/scala/chimp/client/McpClient.scala @@ -106,6 +106,12 @@ trait McpClient[F[_]]: /** Fulfils the input a task is waiting for while it is `InputRequired` (MCP Tasks extension, experimental). */ def updateTask(taskId: String, inputResponses: Json): F[Unit] + /** Invokes a tool, declaring support for the Tasks extension (experimental). The server may answer either directly with a + * [[chimp.protocol.CallToolResult]] (`Left`) or, for a long-running call, with a [[chimp.protocol.CreateTaskResult]] task handle + * (`Right`) that is then driven with [[getTask]] / [[cancelTask]] / [[updateTask]]. + */ + def callToolWithTasks(name: String, arguments: Json): F[Either[CallToolResult, CreateTaskResult]] + /** An [[McpClient]] used over a [[chimp.client.transport.ClientBidirectionalTransport]], which additionally supports server-initiated * interactions: subscribing to resource updates, notifying the server about changes to the client's roots, and handling notifications * pushed by the server. diff --git a/client/src/main/scala/chimp/client/McpClientImpl.scala b/client/src/main/scala/chimp/client/McpClientImpl.scala index 9c79d49..adc0b9b 100644 --- a/client/src/main/scala/chimp/client/McpClientImpl.scala +++ b/client/src/main/scala/chimp/client/McpClientImpl.scala @@ -229,6 +229,20 @@ object McpClientImpl: override def updateTask(taskId: String, inputResponses: Json): F[Unit] = sendRequest[Json]("tasks/update", Some(UpdateTaskParams(taskId, inputResponses).asJson)).map(_ => ()) + override def callToolWithTasks(name: String, arguments: Json): F[Either[CallToolResult, CreateTaskResult]] = + requireServerCapability("tools/call", _.tools.isDefined): + val params = CallToolParams(name = name, arguments = arguments, _meta = Some(TasksExtension.clientCapabilityMeta)).asJson + sendRequest[Json]("tools/call", Some(params)).flatMap: json => + val isTask = json.hcursor.downField("resultType").as[String].toOption.contains("task") + if isTask then + json.as[CreateTaskResult] match + case Right(task) => monad.unit(Right(task)) + case Left(error) => monad.error(McpProtocolException(s"Failed to decode CreateTaskResult: ${error.getMessage}")) + else + json.as[CallToolResult] match + case Right(result) => monad.unit(Left(result)) + case Left(error) => monad.error(McpProtocolException(s"Failed to decode CallToolResult: ${error.getMessage}")) + protected def requireServerCapability[A](method: String, present: ServerCapabilities => Boolean)(action: => F[A]): F[A] = if present(serverCapabilities) then action else monad.error(McpProtocolException(s"Server did not negotiate the capability required for $method")) diff --git a/client/src/test/scala/chimp/client/TasksClientSpec.scala b/client/src/test/scala/chimp/client/TasksClientSpec.scala index bc33063..5c70e38 100644 --- a/client/src/test/scala/chimp/client/TasksClientSpec.scala +++ b/client/src/test/scala/chimp/client/TasksClientSpec.scala @@ -63,3 +63,22 @@ class TasksClientSpec extends AnyFlatSpec with Matchers: .thenRespondAdjust("", StatusCode.Accepted) noException should be thrownBy client(backend).cancelTask("t1") + + it should "declare task support and parse a task handle from callToolWithTasks" in: + val created = CreateTaskResult(taskId = "t9", status = TaskStatus.Working) + val createdEnvelope = (JSONRPCMessage.Response(id = RequestId(1), result = created.asJson): JSONRPCMessage).asJson.noSpaces + val initResult = InitializeResult( + protocolVersion = ProtocolVersion.Latest.name, + capabilities = ServerCapabilities(tools = Some(ServerToolsCapability())), + serverInfo = Implementation(name = "s", version = "1") + ) + val toolsInitEnvelope = (JSONRPCMessage.Response(id = RequestId(1), result = initResult.asJson): JSONRPCMessage).asJson.noSpaces + val backend = SyncBackendStub + .whenRequestMatches(envelopeFor("initialize", _)) + .thenRespondAdjust(toolsInitEnvelope) + .whenRequestMatches(req => envelopeFor("tools/call", req) && envelopeFor(TasksExtension.Id, req)) + .thenRespondAdjust(createdEnvelope) + .whenAnyRequest + .thenRespondAdjust("", StatusCode.Accepted) + + client(backend).callToolWithTasks("slow", io.circe.Json.obj()) shouldBe Right(created) diff --git a/core/src/main/scala/chimp/protocol/JsonRpc.scala b/core/src/main/scala/chimp/protocol/JsonRpc.scala index 95f04d3..aeae83e 100644 --- a/core/src/main/scala/chimp/protocol/JsonRpc.scala +++ b/core/src/main/scala/chimp/protocol/JsonRpc.scala @@ -67,3 +67,4 @@ enum JSONRPCErrorCodes(val code: Int): case InternalError extends JSONRPCErrorCodes(-32603) case InvocationError extends JSONRPCErrorCodes(-32000) case ResourceNotFound extends JSONRPCErrorCodes(-32002) + case MissingRequiredClientCapability extends JSONRPCErrorCodes(-32003) diff --git a/core/src/main/scala/chimp/protocol/Lifecycle.scala b/core/src/main/scala/chimp/protocol/Lifecycle.scala index ed859a1..a398d3c 100644 --- a/core/src/main/scala/chimp/protocol/Lifecycle.scala +++ b/core/src/main/scala/chimp/protocol/Lifecycle.scala @@ -23,7 +23,8 @@ final case class ServerCapabilities( completions: Option[Json] = None, prompts: Option[ServerPromptsCapability] = None, resources: Option[ServerResourcesCapability] = None, - tools: Option[ServerToolsCapability] = None + tools: Option[ServerToolsCapability] = None, + extensions: Option[Map[String, Json]] = None ) derives Codec final case class InitializeParams( diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index 21d5431..729fd5f 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -9,6 +9,20 @@ import io.circe.{Codec, Decoder, Encoder, Json} object TasksExtension: val Id: String = "io.modelcontextprotocol/tasks" + /** `_meta` key under which a client declares its per-request capabilities. */ + val ClientCapabilitiesMetaKey: String = "io.modelcontextprotocol/clientCapabilities" + + /** The `_meta` entry a client adds to a request to declare support for the Tasks extension, so the server may answer with a task. */ + def clientCapabilityMeta: Map[String, Json] = + Map(ClientCapabilitiesMetaKey -> Json.obj("extensions" -> Json.obj(Id -> Json.obj()))) + + /** Whether the given request `_meta` declares support for the Tasks extension. */ + def declaredIn(meta: Option[Map[String, Json]]): Boolean = + meta + .flatMap(_.get(ClientCapabilitiesMetaKey)) + .flatMap(_.hcursor.downField("extensions").downField(Id).focus) + .isDefined + /** State of a task. Terminal states are `Completed`, `Failed` and `Cancelled`. */ enum TaskStatus: case Working, InputRequired, Completed, Failed, Cancelled diff --git a/docs/server/capabilities.md b/docs/server/capabilities.md index f4f34cf..f2b8280 100644 --- a/docs/server/capabilities.md +++ b/docs/server/capabilities.md @@ -29,3 +29,28 @@ val server = StreamingMcpServer[Identity]().addStreamingTool(work) ``` Server-wide capabilities are enabled by registering a handler — only what you wire up is advertised: `.withCompletion`, `.withLoggingLevel`, `.withSubscriptions`. + +## Tasks (experimental) + +With the [Tasks extension](https://github.com/modelcontextprotocol/modelcontextprotocol/blob/main/seps/2663-tasks-extension.md) (`io.modelcontextprotocol/tasks`) the server answers a long-running `tools/call` with a durable task handle instead of blocking. The client (which must declare the extension in its per-request `_meta`) then polls with `tasks/get` and collects the result once the task is `completed`. + +Enable it with `.withTasks`, providing a `TaskStore` and a `TaskExecutor` for the effect type. The synchronous server uses a thread-pool executor: + +```scala mdoc:compile-only +import chimp.server.* +import io.circe.Codec +import sttp.monad.{IdentityMonad, MonadError} +import sttp.shared.Identity +import sttp.tapir.Schema + +given MonadError[Identity] = IdentityMonad + +case class ReportInput(days: Int) derives Codec, Schema + +val report = tool("report").input[ReportInput].handle(in => ToolResult.text(s"report for ${in.days} days")) + +val server = McpServer(tools = List(report)) + .withTasks(TaskSupport(TaskStore.inMemory[Identity], TaskExecutor.threadPool())) +``` + +The server runs the tool in the background, transitions the task to `completed` (with the tool's `CallToolResult`) or `failed`, and answers `tasks/cancel` by interrupting the worker. `useTask` and `requireTask` on `TaskSupport` control, per tool, whether a task is offered or required; a required task with no client support fails with `-32003`. Full server-initiated `input_required` flows are not yet implemented. diff --git a/server/src/main/scala/chimp/server/McpHandler.scala b/server/src/main/scala/chimp/server/McpHandler.scala index 585dcee..35ce9f8 100644 --- a/server/src/main/scala/chimp/server/McpHandler.scala +++ b/server/src/main/scala/chimp/server/McpHandler.scala @@ -82,6 +82,12 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD jsonResponse(JSONRPCMessage.Response(id = id, result = ListToolsResponse(toolDefinitions).asJson)).unit case "tools/call" => handleToolsCall(params, id, headers, makeContext).map(jsonResponse) + case "tasks/get" if server.tasks.isDefined => + handleTasksGet(params, id).map(jsonResponse) + case "tasks/cancel" if server.tasks.isDefined => + handleTasksCancel(params, id).map(jsonResponse) + case "tasks/update" if server.tasks.isDefined => + handleTasksUpdate(params, id).map(jsonResponse) case "resources/list" if hasResources => jsonResponse(JSONRPCMessage.Response(id = id, result = ListResourcesResult(server.resources.map(_.definition)).asJson)).unit case "resources/templates/list" if hasResources => @@ -126,7 +132,8 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD prompts = Option.when(server.prompts.nonEmpty)(ServerPromptsCapability(listChanged = Some(false))), resources = Option.when(hasResources)(ServerResourcesCapability(subscribe = Some(server.subscriptions.isDefined), listChanged = Some(false))), - tools = Option.when(server.tools.nonEmpty)(ServerToolsCapability(listChanged = Some(false))) + tools = Option.when(server.tools.nonEmpty)(ServerToolsCapability(listChanged = Some(false))), + extensions = server.tasks.map(_ => Map(TasksExtension.Id -> Json.obj())) ) val result = InitializeResult( protocolVersion = negotiated.name, @@ -142,6 +149,8 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD val name = params.flatMap(_.hcursor.downField("name").as[String].toOption) val arguments = params.flatMap(_.hcursor.downField("arguments").focus).getOrElse(Json.obj()) val progressToken = params.flatMap(_.hcursor.downField("_meta").downField("progressToken").as[ProgressToken].toOption) + val requestMeta = params.flatMap(_.hcursor.downField("_meta").as[Map[String, Json]].toOption) + val clientSupportsTasks = TasksExtension.declaredIn(requestMeta) name match case Some(name) => toolsByName.get(name) match @@ -149,10 +158,20 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD tool.inputDecoder.decodeJson(arguments) match case Right(input) => val context = makeContext(progressToken) - tool - .logic(input, context, headers) - .map: result => - toolCallResponse(id, result) + server.tasks match + case Some(support) if support.requireTask(name) && !clientSupportsTasks => + protocolError( + id, + JSONRPCErrorCodes.MissingRequiredClientCapability.code, + s"Tool '$name' requires the ${TasksExtension.Id} client capability" + ).unit + case Some(support) if clientSupportsTasks && support.useTask(name) => + startTask(support, id, tool, input, context, headers) + case _ => + tool + .logic(input, context, headers) + .map: result => + toolCallResponse(id, result) case Left(decodingError) => val snippet = arguments.noSpaces.take(200) protocolError( @@ -164,19 +183,111 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, "Missing tool name").unit - private def toolCallResponse(id: RequestId, result: ToolResult[?]): JSONRPCMessage = + private def toCallToolResult(result: ToolResult[?]): CallToolResult = // for backwards compatibility, structured output is serialized into a text block, unless the tool returned content of its own val content = result.structuredContent match case Some(json) if result.content.isEmpty => List(ToolContent.Text(text = json.noSpaces)) case _ => result.content - JSONRPCMessage.Response( - id = id, - result = CallToolResult( - content = content, - structuredContent = result.structuredContent, - isError = result.isError - ).asJson + CallToolResult(content = content, structuredContent = result.structuredContent, isError = result.isError) + + private def toolCallResponse(id: RequestId, result: ToolResult[?]): JSONRPCMessage = + JSONRPCMessage.Response(id = id, result = toCallToolResult(result).asJson) + + private def startTask[I]( + support: TaskSupport[F], + id: RequestId, + tool: ServerTool[I, ?, F, C], + input: I, + context: C, + headers: Seq[Header] + )(using m: MonadError[F]): F[JSONRPCMessage] = + val taskId = java.util.UUID.randomUUID().toString + val now = java.time.Instant.now().toString + val initial = GetTaskResult( + taskId = taskId, + status = TaskStatus.Working, + createdAt = Some(now), + lastUpdatedAt = Some(now), + ttlMs = support.ttlMs, + pollIntervalMs = support.pollIntervalMs, + resultType = Some("complete") ) + // handleError takes its body by-name, so a synchronous (Identity) tool that throws is caught here too + val body: () => F[Unit] = () => + m.handleError( + m.flatMap(tool.logic(input, context, headers))(result => + finishTask(support, taskId, TaskStatus.Completed, result = Some(toCallToolResult(result).asJson)) + ) + ) { case t => + finishTask( + support, + taskId, + TaskStatus.Failed, + error = Some(JSONRPCErrorObject(JSONRPCErrorCodes.InternalError.code, Option(t.getMessage).getOrElse("Task failed")).asJson) + ) + } + support.store + .create(initial) + .flatMap(_ => support.executor.start(taskId, body)) + .map: _ => + JSONRPCMessage.Response( + id = id, + result = CreateTaskResult( + taskId = taskId, + status = TaskStatus.Working, + createdAt = Some(now), + lastUpdatedAt = Some(now), + ttlMs = support.ttlMs, + pollIntervalMs = support.pollIntervalMs + ).asJson + ) + + // only transition a task that is still working, so a cancellation is not overwritten by a late completion + private def finishTask( + support: TaskSupport[F], + taskId: String, + status: TaskStatus, + result: Option[Json] = None, + error: Option[Json] = None + )(using + MonadError[F] + ): F[Unit] = + support.store + .update(taskId): current => + if current.status == TaskStatus.Working then + current.copy(status = status, result = result, error = error, lastUpdatedAt = Some(java.time.Instant.now().toString)) + else current + .map(_ => ()) + + private def handleTasksGet(params: Option[Json], id: RequestId)(using MonadError[F]): F[JSONRPCMessage] = + decodeParams[GetTaskParams](params, id): p => + server.tasks.get.store + .get(p.taskId) + .map: + case Some(task) => JSONRPCMessage.Response(id = id, result = task.asJson) + case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}") + + private def handleTasksCancel(params: Option[Json], id: RequestId)(using MonadError[F]): F[JSONRPCMessage] = + decodeParams[CancelTaskParams](params, id): p => + val support = server.tasks.get + support.store + .update(p.taskId): current => + if TaskStatus.isTerminal(current.status) then current + else current.copy(status = TaskStatus.Cancelled, lastUpdatedAt = Some(java.time.Instant.now().toString)) + .flatMap: + case Some(_) => support.executor.cancel(p.taskId).map(_ => taskAck(id, p.taskId, TaskStatus.Cancelled)) + case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}").unit + + private def handleTasksUpdate(params: Option[Json], id: RequestId)(using MonadError[F]): F[JSONRPCMessage] = + decodeParams[UpdateTaskParams](params, id): p => + server.tasks.get.store + .get(p.taskId) + .map: + case Some(task) => taskAck(id, task.taskId, task.status) + case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}") + + private def taskAck(id: RequestId, taskId: String, status: TaskStatus): JSONRPCMessage = + JSONRPCMessage.Response(id = id, result = TaskAck(taskId = Some(taskId), status = Some(status)).asJson) private def handleResourcesRead(params: Option[Json], id: RequestId, headers: Seq[Header])(using MonadError[F]): F[JSONRPCMessage] = decodeParams[ReadResourceParams](params, id): params => diff --git a/server/src/main/scala/chimp/server/McpServer.scala b/server/src/main/scala/chimp/server/McpServer.scala index 75fca77..f8eb9a2 100644 --- a/server/src/main/scala/chimp/server/McpServer.scala +++ b/server/src/main/scala/chimp/server/McpServer.scala @@ -26,6 +26,7 @@ sealed trait McpServerDef[F[_], C <: ServerContext[F]]: def completion: Option[CompletionHandler[F]] def loggingLevel: Option[SetLoggingLevelHandler[F]] def subscriptions: Option[ResourceSubscriptions[F]] + def tasks: Option[TaskSupport[F]] case class McpServer[F[_]]( name: String = "Chimp MCP server", @@ -39,7 +40,8 @@ case class McpServer[F[_]]( resourceTemplates: List[ServerResourceTemplate[F]] = Nil, completion: Option[CompletionHandler[F]] = None, loggingLevel: Option[SetLoggingLevelHandler[F]] = None, - subscriptions: Option[ResourceSubscriptions[F]] = None + subscriptions: Option[ResourceSubscriptions[F]] = None, + tasks: Option[TaskSupport[F]] = None ) extends McpServerDef[F, ServerContext[F]]: def name(value: String): McpServer[F] = copy(name = value) @@ -89,6 +91,9 @@ case class McpServer[F[_]]( def withSubscriptions(handler: ResourceSubscriptions[F]): McpServer[F] = copy(subscriptions = Some(handler)) + def withTasks(support: TaskSupport[F]): McpServer[F] = + copy(tasks = Some(support)) + def endpoint(path: List[String]): ServerEndpoint[Any, F] = ServerHttpTransport(path).serve(this) def streaming: StreamingMcpServer[F] = @@ -104,7 +109,8 @@ case class McpServer[F[_]]( resourceTemplates, completion, loggingLevel, - subscriptions + subscriptions, + tasks ) case class StreamingMcpServer[F[_]]( @@ -119,7 +125,8 @@ case class StreamingMcpServer[F[_]]( resourceTemplates: List[ServerResourceTemplate[F]] = Nil, completion: Option[CompletionHandler[F]] = None, loggingLevel: Option[SetLoggingLevelHandler[F]] = None, - subscriptions: Option[ResourceSubscriptions[F]] = None + subscriptions: Option[ResourceSubscriptions[F]] = None, + tasks: Option[TaskSupport[F]] = None ) extends McpServerDef[F, StreamingServerContext[F]]: def name(value: String): StreamingMcpServer[F] = copy(name = value) @@ -174,3 +181,6 @@ case class StreamingMcpServer[F[_]]( def withSubscriptions(handler: ResourceSubscriptions[F]): StreamingMcpServer[F] = copy(subscriptions = Some(handler)) + + def withTasks(support: TaskSupport[F]): StreamingMcpServer[F] = + copy(tasks = Some(support)) diff --git a/server/src/main/scala/chimp/server/TaskSupport.scala b/server/src/main/scala/chimp/server/TaskSupport.scala new file mode 100644 index 0000000..a96bd68 --- /dev/null +++ b/server/src/main/scala/chimp/server/TaskSupport.scala @@ -0,0 +1,74 @@ +package chimp.server + +import chimp.protocol.GetTaskResult +import sttp.monad.MonadError +import sttp.shared.Identity + +import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Executors, Future as JavaFuture} + +/** Durable-ish store of task state for the Tasks extension, addressable by task id. The default in-memory implementation keeps tasks for + * the lifetime of the process. + */ +trait TaskStore[F[_]]: + def create(task: GetTaskResult): F[Unit] + def get(taskId: String): F[Option[GetTaskResult]] + + /** Applies `f` to the stored task if present, atomically, and returns the updated task. */ + def update(taskId: String)(f: GetTaskResult => GetTaskResult): F[Option[GetTaskResult]] + +object TaskStore: + def inMemory[F[_]](using m: MonadError[F]): TaskStore[F] = new TaskStore[F]: + private val tasks = ConcurrentHashMap[String, GetTaskResult]() + + def create(task: GetTaskResult): F[Unit] = m.eval: + val _ = tasks.put(task.taskId, task) + () + + def get(taskId: String): F[Option[GetTaskResult]] = m.eval(Option(tasks.get(taskId))) + + def update(taskId: String)(f: GetTaskResult => GetTaskResult): F[Option[GetTaskResult]] = m.eval: + Option(tasks.computeIfPresent(taskId, (_, current) => f(current))) + +/** Runs task bodies in the background and supports best-effort cancellation. The body is passed as a thunk so that, on eager effect types + * such as `Identity`, it is only run on the background worker rather than at the call site. + */ +trait TaskExecutor[F[_]]: + def start(taskId: String, body: () => F[Unit]): F[Unit] + def cancel(taskId: String): F[Unit] + +object TaskExecutor: + + /** A thread-pool executor for synchronous (`Identity`) servers, such as the Netty sync server. Cancellation interrupts the worker thread. + */ + def threadPool(pool: ExecutorService = Executors.newCachedThreadPool()): TaskExecutor[Identity] = new TaskExecutor[Identity]: + private val running = ConcurrentHashMap[String, JavaFuture[?]]() + + def start(taskId: String, body: () => Identity[Unit]): Identity[Unit] = + val future = pool.submit(new Runnable: + def run(): Unit = + try body() + finally + val _ = running.remove(taskId)) + val _ = running.put(taskId, future) + () + + def cancel(taskId: String): Identity[Unit] = + val _ = Option(running.remove(taskId)).foreach(_.cancel(true)) + () + +/** Bundles everything a server needs to answer requests with tasks (Tasks extension, experimental). + * + * @param useTask + * Given a tool name, whether to answer its `tools/call` with a task when the client declares task support. Defaults to always. + * @param requireTask + * Given a tool name, whether a task is required; if the client does not declare task support, the call fails with `-32003`. Defaults to + * never. + */ +final case class TaskSupport[F[_]]( + store: TaskStore[F], + executor: TaskExecutor[F], + ttlMs: Option[Long] = Some(3600000L), + pollIntervalMs: Option[Long] = Some(1000L), + useTask: String => Boolean = (_: String) => true, + requireTask: String => Boolean = (_: String) => false +) diff --git a/server/src/test/scala/chimp/server/TaskServerSpec.scala b/server/src/test/scala/chimp/server/TaskServerSpec.scala new file mode 100644 index 0000000..1b912f0 --- /dev/null +++ b/server/src/test/scala/chimp/server/TaskServerSpec.scala @@ -0,0 +1,127 @@ +package chimp.server + +import chimp.protocol.* +import chimp.protocol.JSONRPCMessage.given +import io.circe.Json +import io.circe.syntax.* +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import sttp.monad.{IdentityMonad, MonadError} +import sttp.shared.Identity +import sttp.tapir.Schema + +class TaskServerSpec extends AnyFlatSpec with Matchers: + import JSONRPCMessage.* + + private given MonadError[Identity] = IdentityMonad + + case class TIn(message: String) derives Schema, io.circe.Codec + + private val instant = tool("instant").input[TIn].handle(in => ToolResult.text(s"echo:${in.message}")) + private val slow = tool("slow") + .input[TIn] + .handle: in => + Thread.sleep(150) + ToolResult.text(s"slow:${in.message}") + private val boom = tool("boom").input[TIn].handle(_ => throw new RuntimeException("boom")) + private val forever = tool("forever") + .input[TIn] + .handle: _ => + Thread.sleep(10000) + ToolResult.text("done") + + private def handlerWith(requireTask: String => Boolean = _ => false): McpHandler[Identity, ServerContext[Identity]] = + val support = TaskSupport[Identity]( + store = TaskStore.inMemory[Identity], + executor = TaskExecutor.threadPool(), + requireTask = requireTask + ) + McpHandler(McpServer(tools = List(instant, slow, boom, forever)).withTasks(support)) + + private def resultJson(response: McpResponse): Json = + val json = response match + case McpResponse.JsonResponse(j) => j + case McpResponse.EmptyAcceptResponse => fail("expected JsonResponse") + json.as[JSONRPCMessage].getOrElse(fail("decode JSONRPCMessage")) match + case Response(_, _, result) => result + case other => fail(s"expected Response, got $other") + + private def errorObj(response: McpResponse): JSONRPCErrorObject = + val json = response match + case McpResponse.JsonResponse(j) => j + case McpResponse.EmptyAcceptResponse => fail("expected JsonResponse") + json.as[JSONRPCMessage].getOrElse(fail("decode JSONRPCMessage")) match + case Error(_, _, err) => err + case other => fail(s"expected Error, got $other") + + private def callToolReq(name: String, withTasks: Boolean): Json = + val meta = Option.when(withTasks)(TasksExtension.clientCapabilityMeta) + val params = CallToolParams(name = name, arguments = TIn("hi").asJson, _meta = meta).asJson + (Request(method = "tools/call", params = Some(params), id = RequestId("call")): JSONRPCMessage).asJson + + private def pollTask(handler: McpHandler[Identity, ServerContext[Identity]], taskId: String): GetTaskResult = + var last = GetTaskResult(taskId = taskId, status = TaskStatus.Working) + var done = false + var i = 0 + while !done && i < 200 do + val req = (Request(method = "tasks/get", params = Some(GetTaskParams(taskId).asJson), id = RequestId("get")): JSONRPCMessage).asJson + last = resultJson(handler.handleJsonRpc(req, Seq.empty)).as[GetTaskResult].getOrElse(fail("decode GetTaskResult")) + if TaskStatus.isTerminal(last.status) then done = true + else + Thread.sleep(20) + i += 1 + last + + "a task-enabled server" should "run tools/call synchronously when the client does not declare task support" in: + val handler = handlerWith() + val result = resultJson(handler.handleJsonRpc(callToolReq("instant", withTasks = false), Seq.empty)) + val call = result.as[CallToolResult].getOrElse(fail("decode CallToolResult")) + call.content.head shouldBe ToolContent.Text("text", "echo:hi") + + it should "answer with a task and complete it when the client declares support" in: + val handler = handlerWith() + val created = resultJson(handler.handleJsonRpc(callToolReq("slow", withTasks = true), Seq.empty)) + .as[CreateTaskResult] + .getOrElse(fail("decode CreateTaskResult")) + created.status shouldBe TaskStatus.Working + created.taskId should not be empty + + val finished = pollTask(handler, created.taskId) + finished.status shouldBe TaskStatus.Completed + finished.result + .flatMap(_.as[CallToolResult].toOption) + .map(_.content.head) shouldBe Some(ToolContent.Text("text", "slow:hi")) + + it should "report a failed task when the tool throws" in: + val handler = handlerWith() + val created = resultJson(handler.handleJsonRpc(callToolReq("boom", withTasks = true), Seq.empty)) + .as[CreateTaskResult] + .getOrElse(fail("decode CreateTaskResult")) + + val finished = pollTask(handler, created.taskId) + finished.status shouldBe TaskStatus.Failed + finished.error.isDefined shouldBe true + + it should "cancel a running task" in: + val handler = handlerWith() + val created = resultJson(handler.handleJsonRpc(callToolReq("forever", withTasks = true), Seq.empty)) + .as[CreateTaskResult] + .getOrElse(fail("decode CreateTaskResult")) + + val cancelReq = + (Request(method = "tasks/cancel", params = Some(CancelTaskParams(created.taskId).asJson), id = RequestId("c")): JSONRPCMessage).asJson + val ack = resultJson(handler.handleJsonRpc(cancelReq, Seq.empty)).as[TaskAck].getOrElse(fail("decode TaskAck")) + ack.status shouldBe Some(TaskStatus.Cancelled) + + pollTask(handler, created.taskId).status shouldBe TaskStatus.Cancelled + + it should "reject a required-task call that lacks the client capability with -32003" in: + val handler = handlerWith(requireTask = _ == "slow") + val err = errorObj(handler.handleJsonRpc(callToolReq("slow", withTasks = false), Seq.empty)) + err.code shouldBe JSONRPCErrorCodes.MissingRequiredClientCapability.code + + it should "advertise the tasks extension capability on initialize" in: + val handler = handlerWith() + val req = (Request(method = "initialize", id = RequestId("i")): JSONRPCMessage).asJson + val result = resultJson(handler.handleJsonRpc(req, Seq.empty)).as[InitializeResult].getOrElse(fail("decode InitializeResult")) + result.capabilities.extensions.map(_.keySet) shouldBe Some(Set(TasksExtension.Id)) From c5ac0d30609b0c616444353206bd17b6b3104c74 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Mon, 31 Aug 2026 19:42:45 +0200 Subject: [PATCH 03/15] refactor: use domain types instead of stringly-typed fields in Tasks Replace primitive-typed task fields with domain types, keeping the exact wire format: - taskId: opaque TaskId instead of String, throughout core, client and server. - createdAt / lastUpdatedAt: java.time.Instant instead of String (ISO 8601). - ttl / pollInterval: FiniteDuration instead of Long; still serialized as the integer-millisecond wire fields ttlMs / pollIntervalMs via Codec.forProductN. - GetTaskResult.error: JSONRPCErrorObject instead of raw Json. - inputRequests / UpdateTaskParams.inputResponses: Map[String, Json] instead of raw Json. - TaskSupport.ttl / pollInterval: FiniteDuration; TaskStore / TaskExecutor keyed by TaskId. Co-Authored-By: Claude Opus 4.8 --- .../main/scala/chimp/client/McpClient.scala | 6 +- .../scala/chimp/client/McpClientImpl.scala | 6 +- .../scala/chimp/client/TasksClientSpec.scala | 10 +-- .../src/main/scala/chimp/protocol/Tasks.scala | 90 ++++++++++++++----- .../test/scala/chimp/protocol/TasksSpec.scala | 18 +++- docs/client/capabilities.md | 2 +- .../main/scala/chimp/server/McpHandler.scala | 24 ++--- .../main/scala/chimp/server/TaskSupport.scala | 27 +++--- .../scala/chimp/server/TaskServerSpec.scala | 4 +- 9 files changed, 124 insertions(+), 63 deletions(-) diff --git a/client/src/main/scala/chimp/client/McpClient.scala b/client/src/main/scala/chimp/client/McpClient.scala index ceef2f1..60b4d1f 100644 --- a/client/src/main/scala/chimp/client/McpClient.scala +++ b/client/src/main/scala/chimp/client/McpClient.scala @@ -98,13 +98,13 @@ trait McpClient[F[_]]: * [[chimp.protocol.CreateTaskResult]], the returned `taskId` is polled with this method until the task reaches a terminal state; the * underlying result is then available in [[chimp.protocol.GetTaskResult.result]]. */ - def getTask(taskId: String): F[GetTaskResult] + def getTask(taskId: TaskId): F[GetTaskResult] /** Requests cancellation of a task by its id (MCP Tasks extension, experimental). */ - def cancelTask(taskId: String): F[Unit] + def cancelTask(taskId: TaskId): F[Unit] /** Fulfils the input a task is waiting for while it is `InputRequired` (MCP Tasks extension, experimental). */ - def updateTask(taskId: String, inputResponses: Json): F[Unit] + def updateTask(taskId: TaskId, inputResponses: Map[String, Json]): F[Unit] /** Invokes a tool, declaring support for the Tasks extension (experimental). The server may answer either directly with a * [[chimp.protocol.CallToolResult]] (`Left`) or, for a long-running call, with a [[chimp.protocol.CreateTaskResult]] task handle diff --git a/client/src/main/scala/chimp/client/McpClientImpl.scala b/client/src/main/scala/chimp/client/McpClientImpl.scala index adc0b9b..d52a3fa 100644 --- a/client/src/main/scala/chimp/client/McpClientImpl.scala +++ b/client/src/main/scala/chimp/client/McpClientImpl.scala @@ -220,13 +220,13 @@ object McpClientImpl: val params = ProgressParams(progressToken = token, progress = progress, total = total, message = message).asJson sendNotification("notifications/progress", Some(params)) - override def getTask(taskId: String): F[GetTaskResult] = + override def getTask(taskId: TaskId): F[GetTaskResult] = sendRequest[GetTaskResult]("tasks/get", Some(GetTaskParams(taskId).asJson)) - override def cancelTask(taskId: String): F[Unit] = + override def cancelTask(taskId: TaskId): F[Unit] = sendRequest[Json]("tasks/cancel", Some(CancelTaskParams(taskId).asJson)).map(_ => ()) - override def updateTask(taskId: String, inputResponses: Json): F[Unit] = + override def updateTask(taskId: TaskId, inputResponses: Map[String, Json]): F[Unit] = sendRequest[Json]("tasks/update", Some(UpdateTaskParams(taskId, inputResponses).asJson)).map(_ => ()) override def callToolWithTasks(name: String, arguments: Json): F[Either[CallToolResult, CreateTaskResult]] = diff --git a/client/src/test/scala/chimp/client/TasksClientSpec.scala b/client/src/test/scala/chimp/client/TasksClientSpec.scala index 5c70e38..04d8c4e 100644 --- a/client/src/test/scala/chimp/client/TasksClientSpec.scala +++ b/client/src/test/scala/chimp/client/TasksClientSpec.scala @@ -33,7 +33,7 @@ class TasksClientSpec extends AnyFlatSpec with Matchers: it should "poll a task with tasks/get and expose the underlying result" in: val task = GetTaskResult( - taskId = "t1", + taskId = TaskId("t1"), status = TaskStatus.Completed, result = Some(CallToolResult(content = List(ToolContent.Text(text = "done"))).asJson), resultType = Some("complete") @@ -47,12 +47,12 @@ class TasksClientSpec extends AnyFlatSpec with Matchers: .whenAnyRequest .thenRespondAdjust("", StatusCode.Accepted) - val res = client(backend).getTask("t1") + val res = client(backend).getTask(TaskId("t1")) res.status shouldBe TaskStatus.Completed res.result.flatMap(_.as[CallToolResult].toOption).map(_.content.head) shouldBe Some(ToolContent.Text("text", "done")) it should "cancel a task with tasks/cancel" in: - val ack = TaskAck(taskId = Some("t1"), status = Some(TaskStatus.Cancelled)) + val ack = TaskAck(taskId = Some(TaskId("t1")), status = Some(TaskStatus.Cancelled)) val ackEnvelope = (JSONRPCMessage.Response(id = RequestId(1), result = ack.asJson): JSONRPCMessage).asJson.noSpaces val backend = SyncBackendStub .whenRequestMatches(envelopeFor("initialize", _)) @@ -62,10 +62,10 @@ class TasksClientSpec extends AnyFlatSpec with Matchers: .whenAnyRequest .thenRespondAdjust("", StatusCode.Accepted) - noException should be thrownBy client(backend).cancelTask("t1") + noException should be thrownBy client(backend).cancelTask(TaskId("t1")) it should "declare task support and parse a task handle from callToolWithTasks" in: - val created = CreateTaskResult(taskId = "t9", status = TaskStatus.Working) + val created = CreateTaskResult(taskId = TaskId("t9"), status = TaskStatus.Working) val createdEnvelope = (JSONRPCMessage.Response(id = RequestId(1), result = created.asJson): JSONRPCMessage).asJson.noSpaces val initResult = InitializeResult( protocolVersion = ProtocolVersion.Latest.name, diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index 729fd5f..bc54544 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -2,6 +2,20 @@ package chimp.protocol import io.circe.{Codec, Decoder, Encoder, Json} +import java.time.Instant +import scala.concurrent.duration.{DurationLong, FiniteDuration} + +// the wire encodes durations as an integer number of milliseconds; file-private so it does not leak into the wider protocol scope +private given Codec[FiniteDuration] = + Codec.from(Decoder.decodeLong.map(_.millis), Encoder.encodeLong.contramap(_.toMillis)) + +/** Identifier of a task, generated by the receiver with enough entropy to prevent enumeration. */ +opaque type TaskId = String +object TaskId: + def apply(value: String): TaskId = value + extension (taskId: TaskId) def value: String = taskId + given Codec[TaskId] = Codec.from(Decoder.decodeString, Encoder.encodeString) + /** The MCP Tasks extension (SEP-2663, identifier `io.modelcontextprotocol/tasks`): durable handles that let a receiver answer a request * with a task, which the requestor then polls and later collects the result of. Experimental; the wire format follows the reference * extension and may change. @@ -46,48 +60,84 @@ object TaskStatus: /** Result returned when a receiver answers a request with a task instead of the request's normal result. */ final case class CreateTaskResult( - taskId: String, + taskId: TaskId, status: TaskStatus, - createdAt: Option[String] = None, - lastUpdatedAt: Option[String] = None, - ttlMs: Option[Long] = None, - pollIntervalMs: Option[Long] = None, + createdAt: Option[Instant] = None, + lastUpdatedAt: Option[Instant] = None, + ttl: Option[FiniteDuration] = None, + pollInterval: Option[FiniteDuration] = None, statusMessage: Option[String] = None, resultType: String = "task", _meta: Option[Map[String, Json]] = None -) derives Codec +) -final case class GetTaskParams(taskId: String, _meta: Option[Map[String, Json]] = None) derives Codec +object CreateTaskResult: + given Codec[CreateTaskResult] = + Codec.forProduct9("taskId", "status", "createdAt", "lastUpdatedAt", "ttlMs", "pollIntervalMs", "statusMessage", "resultType", "_meta")( + CreateTaskResult.apply + )(r => (r.taskId, r.status, r.createdAt, r.lastUpdatedAt, r.ttl, r.pollInterval, r.statusMessage, r.resultType, r._meta)) + +final case class GetTaskParams(taskId: TaskId, _meta: Option[Map[String, Json]] = None) derives Codec final case class GetTaskRequest(method: String = "tasks/get", params: GetTaskParams) derives Codec /** Detailed task state returned by `tasks/get`. `result` is present once the task is `Completed`, `error` once it has `Failed`, and - * `inputRequests` while it is `InputRequired`. `result` and `inputRequests` are left as raw JSON, since their shape depends on the request - * the task stands for. + * `inputRequests` while it is `InputRequired`. `result` is left as raw JSON, since its shape depends on the request the task stands for. */ final case class GetTaskResult( - taskId: String, + taskId: TaskId, status: TaskStatus, - createdAt: Option[String] = None, - lastUpdatedAt: Option[String] = None, - ttlMs: Option[Long] = None, - pollIntervalMs: Option[Long] = None, + createdAt: Option[Instant] = None, + lastUpdatedAt: Option[Instant] = None, + ttl: Option[FiniteDuration] = None, + pollInterval: Option[FiniteDuration] = None, statusMessage: Option[String] = None, result: Option[Json] = None, - error: Option[Json] = None, - inputRequests: Option[Json] = None, + error: Option[JSONRPCErrorObject] = None, + inputRequests: Option[Map[String, Json]] = None, resultType: Option[String] = None, _meta: Option[Map[String, Json]] = None -) derives Codec +) + +object GetTaskResult: + given Codec[GetTaskResult] = Codec.forProduct12( + "taskId", + "status", + "createdAt", + "lastUpdatedAt", + "ttlMs", + "pollIntervalMs", + "statusMessage", + "result", + "error", + "inputRequests", + "resultType", + "_meta" + )(GetTaskResult.apply)(r => + ( + r.taskId, + r.status, + r.createdAt, + r.lastUpdatedAt, + r.ttl, + r.pollInterval, + r.statusMessage, + r.result, + r.error, + r.inputRequests, + r.resultType, + r._meta + ) + ) -final case class CancelTaskParams(taskId: String, _meta: Option[Map[String, Json]] = None) derives Codec +final case class CancelTaskParams(taskId: TaskId, _meta: Option[Map[String, Json]] = None) derives Codec final case class CancelTaskRequest(method: String = "tasks/cancel", params: CancelTaskParams) derives Codec -final case class UpdateTaskParams(taskId: String, inputResponses: Json, _meta: Option[Map[String, Json]] = None) derives Codec +final case class UpdateTaskParams(taskId: TaskId, inputResponses: Map[String, Json], _meta: Option[Map[String, Json]] = None) derives Codec final case class UpdateTaskRequest(method: String = "tasks/update", params: UpdateTaskParams) derives Codec /** Acknowledgement returned by `tasks/cancel` and `tasks/update`. */ final case class TaskAck( - taskId: Option[String] = None, + taskId: Option[TaskId] = None, status: Option[TaskStatus] = None, resultType: String = "complete", _meta: Option[Map[String, Json]] = None diff --git a/core/src/test/scala/chimp/protocol/TasksSpec.scala b/core/src/test/scala/chimp/protocol/TasksSpec.scala index 2049501..a226c21 100644 --- a/core/src/test/scala/chimp/protocol/TasksSpec.scala +++ b/core/src/test/scala/chimp/protocol/TasksSpec.scala @@ -6,6 +6,9 @@ import io.circe.syntax.* import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers +import java.time.Instant +import scala.concurrent.duration.* + class TasksSpec extends AnyFlatSpec with Matchers: it should "decode a CreateTaskResult from the spec example" in: @@ -22,10 +25,17 @@ class TasksSpec extends AnyFlatSpec with Matchers: } """ val res = decode[CreateTaskResult](json) - res.map(_.taskId) shouldBe Right("786512e2-9e0d-44bd-8f29-789f320fe840") + res.map(_.taskId) shouldBe Right(TaskId("786512e2-9e0d-44bd-8f29-789f320fe840")) res.map(_.status) shouldBe Right(TaskStatus.Working) - res.map(_.ttlMs) shouldBe Right(Some(3600000L)) - res.map(_.pollIntervalMs) shouldBe Right(Some(5000L)) + res.map(_.createdAt) shouldBe Right(Some(Instant.parse("2025-11-25T10:30:00Z"))) + res.map(_.ttl) shouldBe Right(Some(1.hour)) + res.map(_.pollInterval) shouldBe Right(Some(5.seconds)) + + it should "encode durations as integer milliseconds on the wire" in: + val created = CreateTaskResult(taskId = TaskId("t"), status = TaskStatus.Working, ttl = Some(1.hour), pollInterval = Some(5.seconds)) + val json = created.asJson + json.hcursor.downField("ttlMs").as[Long] shouldBe Right(3600000L) + json.hcursor.downField("pollIntervalMs").as[Long] shouldBe Right(5000L) it should "encode and decode task status with the spec wire strings" in: (TaskStatus.InputRequired: TaskStatus).asJson shouldBe Json.fromString("input_required") @@ -38,7 +48,7 @@ class TasksSpec extends AnyFlatSpec with Matchers: it should "round-trip a completed GetTaskResult carrying the tool result" in: val toolResult = CallToolResult(content = List(ToolContent.Text(text = "Hello, Luca!"))).asJson val task = GetTaskResult( - taskId = "t1", + taskId = TaskId("t1"), status = TaskStatus.Completed, result = Some(toolResult), resultType = Some("complete") diff --git a/docs/client/capabilities.md b/docs/client/capabilities.md index 27c1a5a..f4e725f 100644 --- a/docs/client/capabilities.md +++ b/docs/client/capabilities.md @@ -53,7 +53,7 @@ import chimp.client.* import chimp.protocol.* import zio.{Task, ZIO} -def awaitResult(client: McpClient[Task], taskId: String): Task[GetTaskResult] = +def awaitResult(client: McpClient[Task], taskId: TaskId): Task[GetTaskResult] = client.getTask(taskId).flatMap { task => if TaskStatus.isTerminal(task.status) then ZIO.succeed(task) else awaitResult(client, taskId) diff --git a/server/src/main/scala/chimp/server/McpHandler.scala b/server/src/main/scala/chimp/server/McpHandler.scala index 35ce9f8..593b383 100644 --- a/server/src/main/scala/chimp/server/McpHandler.scala +++ b/server/src/main/scala/chimp/server/McpHandler.scala @@ -201,15 +201,15 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD context: C, headers: Seq[Header] )(using m: MonadError[F]): F[JSONRPCMessage] = - val taskId = java.util.UUID.randomUUID().toString - val now = java.time.Instant.now().toString + val taskId = TaskId(java.util.UUID.randomUUID().toString) + val now = java.time.Instant.now() val initial = GetTaskResult( taskId = taskId, status = TaskStatus.Working, createdAt = Some(now), lastUpdatedAt = Some(now), - ttlMs = support.ttlMs, - pollIntervalMs = support.pollIntervalMs, + ttl = support.ttl, + pollInterval = support.pollInterval, resultType = Some("complete") ) // handleError takes its body by-name, so a synchronous (Identity) tool that throws is caught here too @@ -223,7 +223,7 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD support, taskId, TaskStatus.Failed, - error = Some(JSONRPCErrorObject(JSONRPCErrorCodes.InternalError.code, Option(t.getMessage).getOrElse("Task failed")).asJson) + error = Some(JSONRPCErrorObject(JSONRPCErrorCodes.InternalError.code, Option(t.getMessage).getOrElse("Task failed"))) ) } support.store @@ -237,25 +237,25 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD status = TaskStatus.Working, createdAt = Some(now), lastUpdatedAt = Some(now), - ttlMs = support.ttlMs, - pollIntervalMs = support.pollIntervalMs + ttl = support.ttl, + pollInterval = support.pollInterval ).asJson ) // only transition a task that is still working, so a cancellation is not overwritten by a late completion private def finishTask( support: TaskSupport[F], - taskId: String, + taskId: TaskId, status: TaskStatus, result: Option[Json] = None, - error: Option[Json] = None + error: Option[JSONRPCErrorObject] = None )(using MonadError[F] ): F[Unit] = support.store .update(taskId): current => if current.status == TaskStatus.Working then - current.copy(status = status, result = result, error = error, lastUpdatedAt = Some(java.time.Instant.now().toString)) + current.copy(status = status, result = result, error = error, lastUpdatedAt = Some(java.time.Instant.now())) else current .map(_ => ()) @@ -273,7 +273,7 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD support.store .update(p.taskId): current => if TaskStatus.isTerminal(current.status) then current - else current.copy(status = TaskStatus.Cancelled, lastUpdatedAt = Some(java.time.Instant.now().toString)) + else current.copy(status = TaskStatus.Cancelled, lastUpdatedAt = Some(java.time.Instant.now())) .flatMap: case Some(_) => support.executor.cancel(p.taskId).map(_ => taskAck(id, p.taskId, TaskStatus.Cancelled)) case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}").unit @@ -286,7 +286,7 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD case Some(task) => taskAck(id, task.taskId, task.status) case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}") - private def taskAck(id: RequestId, taskId: String, status: TaskStatus): JSONRPCMessage = + private def taskAck(id: RequestId, taskId: TaskId, status: TaskStatus): JSONRPCMessage = JSONRPCMessage.Response(id = id, result = TaskAck(taskId = Some(taskId), status = Some(status)).asJson) private def handleResourcesRead(params: Option[Json], id: RequestId, headers: Seq[Header])(using MonadError[F]): F[JSONRPCMessage] = diff --git a/server/src/main/scala/chimp/server/TaskSupport.scala b/server/src/main/scala/chimp/server/TaskSupport.scala index a96bd68..6d92491 100644 --- a/server/src/main/scala/chimp/server/TaskSupport.scala +++ b/server/src/main/scala/chimp/server/TaskSupport.scala @@ -1,49 +1,50 @@ package chimp.server -import chimp.protocol.GetTaskResult +import chimp.protocol.{GetTaskResult, TaskId} import sttp.monad.MonadError import sttp.shared.Identity import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Executors, Future as JavaFuture} +import scala.concurrent.duration.{DurationInt, FiniteDuration} /** Durable-ish store of task state for the Tasks extension, addressable by task id. The default in-memory implementation keeps tasks for * the lifetime of the process. */ trait TaskStore[F[_]]: def create(task: GetTaskResult): F[Unit] - def get(taskId: String): F[Option[GetTaskResult]] + def get(taskId: TaskId): F[Option[GetTaskResult]] /** Applies `f` to the stored task if present, atomically, and returns the updated task. */ - def update(taskId: String)(f: GetTaskResult => GetTaskResult): F[Option[GetTaskResult]] + def update(taskId: TaskId)(f: GetTaskResult => GetTaskResult): F[Option[GetTaskResult]] object TaskStore: def inMemory[F[_]](using m: MonadError[F]): TaskStore[F] = new TaskStore[F]: - private val tasks = ConcurrentHashMap[String, GetTaskResult]() + private val tasks = ConcurrentHashMap[TaskId, GetTaskResult]() def create(task: GetTaskResult): F[Unit] = m.eval: val _ = tasks.put(task.taskId, task) () - def get(taskId: String): F[Option[GetTaskResult]] = m.eval(Option(tasks.get(taskId))) + def get(taskId: TaskId): F[Option[GetTaskResult]] = m.eval(Option(tasks.get(taskId))) - def update(taskId: String)(f: GetTaskResult => GetTaskResult): F[Option[GetTaskResult]] = m.eval: + def update(taskId: TaskId)(f: GetTaskResult => GetTaskResult): F[Option[GetTaskResult]] = m.eval: Option(tasks.computeIfPresent(taskId, (_, current) => f(current))) /** Runs task bodies in the background and supports best-effort cancellation. The body is passed as a thunk so that, on eager effect types * such as `Identity`, it is only run on the background worker rather than at the call site. */ trait TaskExecutor[F[_]]: - def start(taskId: String, body: () => F[Unit]): F[Unit] - def cancel(taskId: String): F[Unit] + def start(taskId: TaskId, body: () => F[Unit]): F[Unit] + def cancel(taskId: TaskId): F[Unit] object TaskExecutor: /** A thread-pool executor for synchronous (`Identity`) servers, such as the Netty sync server. Cancellation interrupts the worker thread. */ def threadPool(pool: ExecutorService = Executors.newCachedThreadPool()): TaskExecutor[Identity] = new TaskExecutor[Identity]: - private val running = ConcurrentHashMap[String, JavaFuture[?]]() + private val running = ConcurrentHashMap[TaskId, JavaFuture[?]]() - def start(taskId: String, body: () => Identity[Unit]): Identity[Unit] = + def start(taskId: TaskId, body: () => Identity[Unit]): Identity[Unit] = val future = pool.submit(new Runnable: def run(): Unit = try body() @@ -52,7 +53,7 @@ object TaskExecutor: val _ = running.put(taskId, future) () - def cancel(taskId: String): Identity[Unit] = + def cancel(taskId: TaskId): Identity[Unit] = val _ = Option(running.remove(taskId)).foreach(_.cancel(true)) () @@ -67,8 +68,8 @@ object TaskExecutor: final case class TaskSupport[F[_]]( store: TaskStore[F], executor: TaskExecutor[F], - ttlMs: Option[Long] = Some(3600000L), - pollIntervalMs: Option[Long] = Some(1000L), + ttl: Option[FiniteDuration] = Some(1.hour), + pollInterval: Option[FiniteDuration] = Some(1.second), useTask: String => Boolean = (_: String) => true, requireTask: String => Boolean = (_: String) => false ) diff --git a/server/src/test/scala/chimp/server/TaskServerSpec.scala b/server/src/test/scala/chimp/server/TaskServerSpec.scala index 1b912f0..f5753b9 100644 --- a/server/src/test/scala/chimp/server/TaskServerSpec.scala +++ b/server/src/test/scala/chimp/server/TaskServerSpec.scala @@ -59,7 +59,7 @@ class TaskServerSpec extends AnyFlatSpec with Matchers: val params = CallToolParams(name = name, arguments = TIn("hi").asJson, _meta = meta).asJson (Request(method = "tools/call", params = Some(params), id = RequestId("call")): JSONRPCMessage).asJson - private def pollTask(handler: McpHandler[Identity, ServerContext[Identity]], taskId: String): GetTaskResult = + private def pollTask(handler: McpHandler[Identity, ServerContext[Identity]], taskId: TaskId): GetTaskResult = var last = GetTaskResult(taskId = taskId, status = TaskStatus.Working) var done = false var i = 0 @@ -84,7 +84,7 @@ class TaskServerSpec extends AnyFlatSpec with Matchers: .as[CreateTaskResult] .getOrElse(fail("decode CreateTaskResult")) created.status shouldBe TaskStatus.Working - created.taskId should not be empty + created.taskId.value should not be empty val finished = pollTask(handler, created.taskId) finished.status shouldBe TaskStatus.Completed From ed835e75f09c810965be212ebfb25194d7649a2c Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Mon, 31 Aug 2026 20:36:04 +0200 Subject: [PATCH 04/15] refactor: model task durations with java.time.Duration Use java.time.Duration for ttl / pollInterval, to match the java.time.Instant already used for the task timestamps. The Tasks spec fixes ttlMs / pollIntervalMs as integer milliseconds (not ISO-8601 durations), so the wire encoding is unchanged: the file-private codec maps Duration to/from milliseconds, verified by a test asserting the integer-millisecond wire fields. Co-Authored-By: Claude Opus 4.8 --- core/src/main/scala/chimp/protocol/Tasks.scala | 18 +++++++++--------- .../test/scala/chimp/protocol/TasksSpec.scala | 15 ++++++++++----- .../main/scala/chimp/server/TaskSupport.scala | 6 +++--- 3 files changed, 22 insertions(+), 17 deletions(-) diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index bc54544..b98a7c7 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -2,12 +2,12 @@ package chimp.protocol import io.circe.{Codec, Decoder, Encoder, Json} -import java.time.Instant -import scala.concurrent.duration.{DurationLong, FiniteDuration} +import java.time.{Duration, Instant} -// the wire encodes durations as an integer number of milliseconds; file-private so it does not leak into the wider protocol scope -private given Codec[FiniteDuration] = - Codec.from(Decoder.decodeLong.map(_.millis), Encoder.encodeLong.contramap(_.toMillis)) +// the Tasks wire format encodes durations as an integer number of milliseconds (ttlMs, pollIntervalMs), not as ISO-8601 strings; this +// file-private codec bridges java.time.Duration to that representation without leaking into the wider protocol scope +private given Codec[Duration] = + Codec.from(Decoder.decodeLong.map(Duration.ofMillis), Encoder.encodeLong.contramap(_.toMillis)) /** Identifier of a task, generated by the receiver with enough entropy to prevent enumeration. */ opaque type TaskId = String @@ -64,8 +64,8 @@ final case class CreateTaskResult( status: TaskStatus, createdAt: Option[Instant] = None, lastUpdatedAt: Option[Instant] = None, - ttl: Option[FiniteDuration] = None, - pollInterval: Option[FiniteDuration] = None, + ttl: Option[Duration] = None, + pollInterval: Option[Duration] = None, statusMessage: Option[String] = None, resultType: String = "task", _meta: Option[Map[String, Json]] = None @@ -88,8 +88,8 @@ final case class GetTaskResult( status: TaskStatus, createdAt: Option[Instant] = None, lastUpdatedAt: Option[Instant] = None, - ttl: Option[FiniteDuration] = None, - pollInterval: Option[FiniteDuration] = None, + ttl: Option[Duration] = None, + pollInterval: Option[Duration] = None, statusMessage: Option[String] = None, result: Option[Json] = None, error: Option[JSONRPCErrorObject] = None, diff --git a/core/src/test/scala/chimp/protocol/TasksSpec.scala b/core/src/test/scala/chimp/protocol/TasksSpec.scala index a226c21..f6dbe1b 100644 --- a/core/src/test/scala/chimp/protocol/TasksSpec.scala +++ b/core/src/test/scala/chimp/protocol/TasksSpec.scala @@ -6,8 +6,7 @@ import io.circe.syntax.* import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -import java.time.Instant -import scala.concurrent.duration.* +import java.time.{Duration, Instant} class TasksSpec extends AnyFlatSpec with Matchers: @@ -28,11 +27,17 @@ class TasksSpec extends AnyFlatSpec with Matchers: res.map(_.taskId) shouldBe Right(TaskId("786512e2-9e0d-44bd-8f29-789f320fe840")) res.map(_.status) shouldBe Right(TaskStatus.Working) res.map(_.createdAt) shouldBe Right(Some(Instant.parse("2025-11-25T10:30:00Z"))) - res.map(_.ttl) shouldBe Right(Some(1.hour)) - res.map(_.pollInterval) shouldBe Right(Some(5.seconds)) + res.map(_.ttl) shouldBe Right(Some(Duration.ofHours(1))) + res.map(_.pollInterval) shouldBe Right(Some(Duration.ofSeconds(5))) it should "encode durations as integer milliseconds on the wire" in: - val created = CreateTaskResult(taskId = TaskId("t"), status = TaskStatus.Working, ttl = Some(1.hour), pollInterval = Some(5.seconds)) + val created = + CreateTaskResult( + taskId = TaskId("t"), + status = TaskStatus.Working, + ttl = Some(Duration.ofHours(1)), + pollInterval = Some(Duration.ofSeconds(5)) + ) val json = created.asJson json.hcursor.downField("ttlMs").as[Long] shouldBe Right(3600000L) json.hcursor.downField("pollIntervalMs").as[Long] shouldBe Right(5000L) diff --git a/server/src/main/scala/chimp/server/TaskSupport.scala b/server/src/main/scala/chimp/server/TaskSupport.scala index 6d92491..39cca4b 100644 --- a/server/src/main/scala/chimp/server/TaskSupport.scala +++ b/server/src/main/scala/chimp/server/TaskSupport.scala @@ -4,8 +4,8 @@ import chimp.protocol.{GetTaskResult, TaskId} import sttp.monad.MonadError import sttp.shared.Identity +import java.time.Duration import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Executors, Future as JavaFuture} -import scala.concurrent.duration.{DurationInt, FiniteDuration} /** Durable-ish store of task state for the Tasks extension, addressable by task id. The default in-memory implementation keeps tasks for * the lifetime of the process. @@ -68,8 +68,8 @@ object TaskExecutor: final case class TaskSupport[F[_]]( store: TaskStore[F], executor: TaskExecutor[F], - ttl: Option[FiniteDuration] = Some(1.hour), - pollInterval: Option[FiniteDuration] = Some(1.second), + ttl: Option[Duration] = Some(Duration.ofHours(1)), + pollInterval: Option[Duration] = Some(Duration.ofSeconds(1)), useTask: String => Boolean = (_: String) => true, requireTask: String => Boolean = (_: String) => false ) From f322c10903192a9e5cf78deb8b1da8a9e491d9ac Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Mon, 31 Aug 2026 22:54:57 +0200 Subject: [PATCH 05/15] feat: serialize task durations as ISO-8601 strings Add reusable ISO-8601 duration codecs (DurationCodecs) for both java.time.Duration and scala.concurrent.duration.FiniteDuration, the latter bridging through java.time.Duration (the JDK's ISO-8601 duration type). Encoding always produces an ISO-8601 string (e.g. "PT1H"); decoding also accepts a bare JSON number read as milliseconds, so chimp still interoperates with peers that send the ext-tasks bare-millisecond ttlMs / pollIntervalMs form. The task ttl / pollInterval fields now serialize as ISO-8601 strings. Co-Authored-By: Claude Opus 4.8 --- .../src/main/scala/chimp/protocol/Tasks.scala | 29 ++++++++++++++--- .../chimp/protocol/DurationCodecsSpec.scala | 31 +++++++++++++++++++ .../test/scala/chimp/protocol/TasksSpec.scala | 12 +++++-- 3 files changed, 64 insertions(+), 8 deletions(-) create mode 100644 core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index b98a7c7..fcd7fb0 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -1,13 +1,32 @@ package chimp.protocol -import io.circe.{Codec, Decoder, Encoder, Json} +import io.circe.{Codec, Decoder, DecodingFailure, Encoder} +import io.circe.Json import java.time.{Duration, Instant} +import scala.concurrent.duration.{FiniteDuration, NANOSECONDS} +import scala.util.Try -// the Tasks wire format encodes durations as an integer number of milliseconds (ttlMs, pollIntervalMs), not as ISO-8601 strings; this -// file-private codec bridges java.time.Duration to that representation without leaking into the wider protocol scope -private given Codec[Duration] = - Codec.from(Decoder.decodeLong.map(Duration.ofMillis), Encoder.encodeLong.contramap(_.toMillis)) +/** ISO-8601 duration codecs, for example `"PT1H"` or `"PT1H30M"`. Encoding always produces an ISO-8601 string. To interoperate with peers + * that send the bare-millisecond form, decoding also accepts a JSON number, read as milliseconds. The `FiniteDuration` codec bridges + * through [[java.time.Duration]], which is the JDK's ISO-8601 duration type. + */ +object DurationCodecs: + private val decodeDuration: Decoder[Duration] = Decoder.instance: c => + c.value.asString match + case Some(iso) => + Try(Duration.parse(iso)).toEither.left + .map(e => DecodingFailure(s"Invalid ISO-8601 duration '$iso': ${e.getMessage}", c.history)) + case None => c.as[Long].map(Duration.ofMillis) + + given Codec[Duration] = Codec.from(decodeDuration, Encoder.encodeString.contramap(_.toString)) + + given Codec[FiniteDuration] = Codec.from( + decodeDuration.map(d => FiniteDuration(d.toNanos, NANOSECONDS)), + Encoder.encodeString.contramap(fd => Duration.ofNanos(fd.toNanos).toString) + ) + +import DurationCodecs.given /** Identifier of a task, generated by the receiver with enough entropy to prevent enumeration. */ opaque type TaskId = String diff --git a/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala b/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala new file mode 100644 index 0000000..5bcdcef --- /dev/null +++ b/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala @@ -0,0 +1,31 @@ +package chimp.protocol + +import chimp.protocol.DurationCodecs.given +import io.circe.Json +import io.circe.parser.decode +import io.circe.syntax.* +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.time.Duration +import scala.concurrent.duration.* + +class DurationCodecsSpec extends AnyFlatSpec with Matchers: + + it should "encode a java.time.Duration as an ISO-8601 string" in: + (Duration.ofMinutes(90): Duration).asJson shouldBe Json.fromString("PT1H30M") + + it should "encode a FiniteDuration as an ISO-8601 string" in: + (5.seconds: FiniteDuration).asJson shouldBe Json.fromString("PT5S") + (90.minutes: FiniteDuration).asJson shouldBe Json.fromString("PT1H30M") + + it should "decode an ISO-8601 string to a FiniteDuration" in: + decode[FiniteDuration]("\"PT5S\"") shouldBe Right(5.seconds) + decode[FiniteDuration]("\"PT1H30M\"") shouldBe Right(90.minutes) + + it should "also decode a bare number as milliseconds, for both duration types" in: + decode[Duration]("1500") shouldBe Right(Duration.ofMillis(1500)) + decode[FiniteDuration]("1500") shouldBe Right(1500.millis) + + it should "reject a malformed duration string" in: + decode[FiniteDuration]("\"not-a-duration\"").isLeft shouldBe true diff --git a/core/src/test/scala/chimp/protocol/TasksSpec.scala b/core/src/test/scala/chimp/protocol/TasksSpec.scala index f6dbe1b..0d90b1b 100644 --- a/core/src/test/scala/chimp/protocol/TasksSpec.scala +++ b/core/src/test/scala/chimp/protocol/TasksSpec.scala @@ -30,7 +30,7 @@ class TasksSpec extends AnyFlatSpec with Matchers: res.map(_.ttl) shouldBe Right(Some(Duration.ofHours(1))) res.map(_.pollInterval) shouldBe Right(Some(Duration.ofSeconds(5))) - it should "encode durations as integer milliseconds on the wire" in: + it should "encode durations as ISO-8601 strings on the wire" in: val created = CreateTaskResult( taskId = TaskId("t"), @@ -39,8 +39,14 @@ class TasksSpec extends AnyFlatSpec with Matchers: pollInterval = Some(Duration.ofSeconds(5)) ) val json = created.asJson - json.hcursor.downField("ttlMs").as[Long] shouldBe Right(3600000L) - json.hcursor.downField("pollIntervalMs").as[Long] shouldBe Right(5000L) + json.hcursor.downField("ttlMs").as[String] shouldBe Right("PT1H") + json.hcursor.downField("pollIntervalMs").as[String] shouldBe Right("PT5S") + + it should "decode durations from an ISO-8601 string" in: + val json = """{ "resultType": "task", "taskId": "t", "status": "working", "ttlMs": "PT2H", "pollIntervalMs": "PT10S" }""" + val res = decode[CreateTaskResult](json) + res.map(_.ttl) shouldBe Right(Some(Duration.ofHours(2))) + res.map(_.pollInterval) shouldBe Right(Some(Duration.ofSeconds(10))) it should "encode and decode task status with the spec wire strings" in: (TaskStatus.InputRequired: TaskStatus).asJson shouldBe Json.fromString("input_required") From ccf8d9c4caf200dcf5676b0418fb35866eb5d601 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Mon, 31 Aug 2026 23:34:48 +0200 Subject: [PATCH 06/15] refactor: keep FiniteDuration in the domain, java.time.Duration only in codecs Use scala.concurrent.duration.FiniteDuration for the task ttl / pollInterval fields and TaskSupport, and expose only a Codec[FiniteDuration]. java.time.Duration is now confined to DurationCodecs, where it is used solely as the JDK's ISO-8601 duration parser and formatter (Duration.parse / Duration.ofNanos(..).toString). Co-Authored-By: Claude Opus 4.8 --- .../src/main/scala/chimp/protocol/Tasks.scala | 41 ++++++++++--------- .../chimp/protocol/DurationCodecsSpec.scala | 7 +--- .../test/scala/chimp/protocol/TasksSpec.scala | 15 +++---- .../main/scala/chimp/server/TaskSupport.scala | 6 +-- 4 files changed, 33 insertions(+), 36 deletions(-) diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index fcd7fb0..1b625b1 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -1,29 +1,30 @@ package chimp.protocol -import io.circe.{Codec, Decoder, DecodingFailure, Encoder} -import io.circe.Json +import io.circe.{Codec, Decoder, DecodingFailure, Encoder, Json} -import java.time.{Duration, Instant} -import scala.concurrent.duration.{FiniteDuration, NANOSECONDS} +import java.time.Instant +import scala.concurrent.duration.{DurationLong, FiniteDuration, NANOSECONDS} import scala.util.Try -/** ISO-8601 duration codecs, for example `"PT1H"` or `"PT1H30M"`. Encoding always produces an ISO-8601 string. To interoperate with peers - * that send the bare-millisecond form, decoding also accepts a JSON number, read as milliseconds. The `FiniteDuration` codec bridges - * through [[java.time.Duration]], which is the JDK's ISO-8601 duration type. +/** ISO-8601 duration codec for [[scala.concurrent.duration.FiniteDuration]], for example `"PT1H"` or `"PT1H30M"`. Encoding always produces + * an ISO-8601 string. To interoperate with peers that send the bare-millisecond form, decoding also accepts a JSON number, read as + * milliseconds. `java.time.Duration` is used internally only, as the JDK's ISO-8601 duration parser and formatter. */ object DurationCodecs: - private val decodeDuration: Decoder[Duration] = Decoder.instance: c => - c.value.asString match - case Some(iso) => - Try(Duration.parse(iso)).toEither.left - .map(e => DecodingFailure(s"Invalid ISO-8601 duration '$iso': ${e.getMessage}", c.history)) - case None => c.as[Long].map(Duration.ofMillis) + private def parseIso(iso: String): Either[String, FiniteDuration] = + Try(java.time.Duration.parse(iso)).toEither.left + .map(e => s"Invalid ISO-8601 duration '$iso': ${e.getMessage}") + .map(d => FiniteDuration(d.toNanos, NANOSECONDS)) - given Codec[Duration] = Codec.from(decodeDuration, Encoder.encodeString.contramap(_.toString)) + private def formatIso(duration: FiniteDuration): String = java.time.Duration.ofNanos(duration.toNanos).toString given Codec[FiniteDuration] = Codec.from( - decodeDuration.map(d => FiniteDuration(d.toNanos, NANOSECONDS)), - Encoder.encodeString.contramap(fd => Duration.ofNanos(fd.toNanos).toString) + Decoder.instance { c => + c.value.asString match + case Some(iso) => parseIso(iso).left.map(message => DecodingFailure(message, c.history)) + case None => c.as[Long].map(_.millis) + }, + Encoder.encodeString.contramap(formatIso) ) import DurationCodecs.given @@ -83,8 +84,8 @@ final case class CreateTaskResult( status: TaskStatus, createdAt: Option[Instant] = None, lastUpdatedAt: Option[Instant] = None, - ttl: Option[Duration] = None, - pollInterval: Option[Duration] = None, + ttl: Option[FiniteDuration] = None, + pollInterval: Option[FiniteDuration] = None, statusMessage: Option[String] = None, resultType: String = "task", _meta: Option[Map[String, Json]] = None @@ -107,8 +108,8 @@ final case class GetTaskResult( status: TaskStatus, createdAt: Option[Instant] = None, lastUpdatedAt: Option[Instant] = None, - ttl: Option[Duration] = None, - pollInterval: Option[Duration] = None, + ttl: Option[FiniteDuration] = None, + pollInterval: Option[FiniteDuration] = None, statusMessage: Option[String] = None, result: Option[Json] = None, error: Option[JSONRPCErrorObject] = None, diff --git a/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala b/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala index 5bcdcef..0b280f7 100644 --- a/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala +++ b/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala @@ -7,14 +7,10 @@ import io.circe.syntax.* import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -import java.time.Duration import scala.concurrent.duration.* class DurationCodecsSpec extends AnyFlatSpec with Matchers: - it should "encode a java.time.Duration as an ISO-8601 string" in: - (Duration.ofMinutes(90): Duration).asJson shouldBe Json.fromString("PT1H30M") - it should "encode a FiniteDuration as an ISO-8601 string" in: (5.seconds: FiniteDuration).asJson shouldBe Json.fromString("PT5S") (90.minutes: FiniteDuration).asJson shouldBe Json.fromString("PT1H30M") @@ -23,8 +19,7 @@ class DurationCodecsSpec extends AnyFlatSpec with Matchers: decode[FiniteDuration]("\"PT5S\"") shouldBe Right(5.seconds) decode[FiniteDuration]("\"PT1H30M\"") shouldBe Right(90.minutes) - it should "also decode a bare number as milliseconds, for both duration types" in: - decode[Duration]("1500") shouldBe Right(Duration.ofMillis(1500)) + it should "also decode a bare number as milliseconds" in: decode[FiniteDuration]("1500") shouldBe Right(1500.millis) it should "reject a malformed duration string" in: diff --git a/core/src/test/scala/chimp/protocol/TasksSpec.scala b/core/src/test/scala/chimp/protocol/TasksSpec.scala index 0d90b1b..805d9f3 100644 --- a/core/src/test/scala/chimp/protocol/TasksSpec.scala +++ b/core/src/test/scala/chimp/protocol/TasksSpec.scala @@ -6,7 +6,8 @@ import io.circe.syntax.* import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers -import java.time.{Duration, Instant} +import java.time.Instant +import scala.concurrent.duration.* class TasksSpec extends AnyFlatSpec with Matchers: @@ -27,16 +28,16 @@ class TasksSpec extends AnyFlatSpec with Matchers: res.map(_.taskId) shouldBe Right(TaskId("786512e2-9e0d-44bd-8f29-789f320fe840")) res.map(_.status) shouldBe Right(TaskStatus.Working) res.map(_.createdAt) shouldBe Right(Some(Instant.parse("2025-11-25T10:30:00Z"))) - res.map(_.ttl) shouldBe Right(Some(Duration.ofHours(1))) - res.map(_.pollInterval) shouldBe Right(Some(Duration.ofSeconds(5))) + res.map(_.ttl) shouldBe Right(Some(1.hour)) + res.map(_.pollInterval) shouldBe Right(Some(5.seconds)) it should "encode durations as ISO-8601 strings on the wire" in: val created = CreateTaskResult( taskId = TaskId("t"), status = TaskStatus.Working, - ttl = Some(Duration.ofHours(1)), - pollInterval = Some(Duration.ofSeconds(5)) + ttl = Some(1.hour), + pollInterval = Some(5.seconds) ) val json = created.asJson json.hcursor.downField("ttlMs").as[String] shouldBe Right("PT1H") @@ -45,8 +46,8 @@ class TasksSpec extends AnyFlatSpec with Matchers: it should "decode durations from an ISO-8601 string" in: val json = """{ "resultType": "task", "taskId": "t", "status": "working", "ttlMs": "PT2H", "pollIntervalMs": "PT10S" }""" val res = decode[CreateTaskResult](json) - res.map(_.ttl) shouldBe Right(Some(Duration.ofHours(2))) - res.map(_.pollInterval) shouldBe Right(Some(Duration.ofSeconds(10))) + res.map(_.ttl) shouldBe Right(Some(2.hours)) + res.map(_.pollInterval) shouldBe Right(Some(10.seconds)) it should "encode and decode task status with the spec wire strings" in: (TaskStatus.InputRequired: TaskStatus).asJson shouldBe Json.fromString("input_required") diff --git a/server/src/main/scala/chimp/server/TaskSupport.scala b/server/src/main/scala/chimp/server/TaskSupport.scala index 39cca4b..6d92491 100644 --- a/server/src/main/scala/chimp/server/TaskSupport.scala +++ b/server/src/main/scala/chimp/server/TaskSupport.scala @@ -4,8 +4,8 @@ import chimp.protocol.{GetTaskResult, TaskId} import sttp.monad.MonadError import sttp.shared.Identity -import java.time.Duration import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Executors, Future as JavaFuture} +import scala.concurrent.duration.{DurationInt, FiniteDuration} /** Durable-ish store of task state for the Tasks extension, addressable by task id. The default in-memory implementation keeps tasks for * the lifetime of the process. @@ -68,8 +68,8 @@ object TaskExecutor: final case class TaskSupport[F[_]]( store: TaskStore[F], executor: TaskExecutor[F], - ttl: Option[Duration] = Some(Duration.ofHours(1)), - pollInterval: Option[Duration] = Some(Duration.ofSeconds(1)), + ttl: Option[FiniteDuration] = Some(1.hour), + pollInterval: Option[FiniteDuration] = Some(1.second), useTask: String => Boolean = (_: String) => true, requireTask: String => Boolean = (_: String) => false ) From 58aac961741c9aee4da4d51aae5cb3ac677261ab Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 00:01:46 +0200 Subject: [PATCH 07/15] fix: encode task durations as integer milliseconds for spec conformance The Tasks extension fixes ttlMs / pollIntervalMs as integer milliseconds, so revert the wire encoding from ISO-8601 back to integer milliseconds. The domain type stays scala.concurrent.duration.FiniteDuration; only the codec changes. Drop the now-unused ISO-8601 DurationCodecs. Co-Authored-By: Claude Opus 4.8 --- .../chimp/protocol/DurationCodecsSpec.scala | 26 ------------------- 1 file changed, 26 deletions(-) delete mode 100644 core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala diff --git a/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala b/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala deleted file mode 100644 index 0b280f7..0000000 --- a/core/src/test/scala/chimp/protocol/DurationCodecsSpec.scala +++ /dev/null @@ -1,26 +0,0 @@ -package chimp.protocol - -import chimp.protocol.DurationCodecs.given -import io.circe.Json -import io.circe.parser.decode -import io.circe.syntax.* -import org.scalatest.flatspec.AnyFlatSpec -import org.scalatest.matchers.should.Matchers - -import scala.concurrent.duration.* - -class DurationCodecsSpec extends AnyFlatSpec with Matchers: - - it should "encode a FiniteDuration as an ISO-8601 string" in: - (5.seconds: FiniteDuration).asJson shouldBe Json.fromString("PT5S") - (90.minutes: FiniteDuration).asJson shouldBe Json.fromString("PT1H30M") - - it should "decode an ISO-8601 string to a FiniteDuration" in: - decode[FiniteDuration]("\"PT5S\"") shouldBe Right(5.seconds) - decode[FiniteDuration]("\"PT1H30M\"") shouldBe Right(90.minutes) - - it should "also decode a bare number as milliseconds" in: - decode[FiniteDuration]("1500") shouldBe Right(1500.millis) - - it should "reject a malformed duration string" in: - decode[FiniteDuration]("\"not-a-duration\"").isLeft shouldBe true From 7d785cf1a114c085c9b1d0f759abdb72fe040f25 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 00:02:27 +0200 Subject: [PATCH 08/15] fix: encode task durations as integer milliseconds for spec conformance The Tasks extension fixes ttlMs / pollIntervalMs as integer milliseconds, so revert the wire encoding from ISO-8601 back to integer milliseconds. The domain type stays scala.concurrent.duration.FiniteDuration; only the codec changes. Co-Authored-By: Claude Opus 4.8 --- .../src/main/scala/chimp/protocol/Tasks.scala | 31 ++++--------------- .../test/scala/chimp/protocol/TasksSpec.scala | 12 ++----- 2 files changed, 9 insertions(+), 34 deletions(-) diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index 1b625b1..a262100 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -1,33 +1,14 @@ package chimp.protocol -import io.circe.{Codec, Decoder, DecodingFailure, Encoder, Json} +import io.circe.{Codec, Decoder, Encoder, Json} import java.time.Instant -import scala.concurrent.duration.{DurationLong, FiniteDuration, NANOSECONDS} -import scala.util.Try +import scala.concurrent.duration.{DurationLong, FiniteDuration} -/** ISO-8601 duration codec for [[scala.concurrent.duration.FiniteDuration]], for example `"PT1H"` or `"PT1H30M"`. Encoding always produces - * an ISO-8601 string. To interoperate with peers that send the bare-millisecond form, decoding also accepts a JSON number, read as - * milliseconds. `java.time.Duration` is used internally only, as the JDK's ISO-8601 duration parser and formatter. - */ -object DurationCodecs: - private def parseIso(iso: String): Either[String, FiniteDuration] = - Try(java.time.Duration.parse(iso)).toEither.left - .map(e => s"Invalid ISO-8601 duration '$iso': ${e.getMessage}") - .map(d => FiniteDuration(d.toNanos, NANOSECONDS)) - - private def formatIso(duration: FiniteDuration): String = java.time.Duration.ofNanos(duration.toNanos).toString - - given Codec[FiniteDuration] = Codec.from( - Decoder.instance { c => - c.value.asString match - case Some(iso) => parseIso(iso).left.map(message => DecodingFailure(message, c.history)) - case None => c.as[Long].map(_.millis) - }, - Encoder.encodeString.contramap(formatIso) - ) - -import DurationCodecs.given +// the Tasks wire format fixes durations as an integer number of milliseconds (ttlMs, pollIntervalMs); this file-private codec keeps the +// FiniteDuration domain type without leaking into the wider protocol scope +private given Codec[FiniteDuration] = + Codec.from(Decoder.decodeLong.map(_.millis), Encoder.encodeLong.contramap(_.toMillis)) /** Identifier of a task, generated by the receiver with enough entropy to prevent enumeration. */ opaque type TaskId = String diff --git a/core/src/test/scala/chimp/protocol/TasksSpec.scala b/core/src/test/scala/chimp/protocol/TasksSpec.scala index 805d9f3..b0665b6 100644 --- a/core/src/test/scala/chimp/protocol/TasksSpec.scala +++ b/core/src/test/scala/chimp/protocol/TasksSpec.scala @@ -31,7 +31,7 @@ class TasksSpec extends AnyFlatSpec with Matchers: res.map(_.ttl) shouldBe Right(Some(1.hour)) res.map(_.pollInterval) shouldBe Right(Some(5.seconds)) - it should "encode durations as ISO-8601 strings on the wire" in: + it should "encode durations as integer milliseconds on the wire" in: val created = CreateTaskResult( taskId = TaskId("t"), @@ -40,14 +40,8 @@ class TasksSpec extends AnyFlatSpec with Matchers: pollInterval = Some(5.seconds) ) val json = created.asJson - json.hcursor.downField("ttlMs").as[String] shouldBe Right("PT1H") - json.hcursor.downField("pollIntervalMs").as[String] shouldBe Right("PT5S") - - it should "decode durations from an ISO-8601 string" in: - val json = """{ "resultType": "task", "taskId": "t", "status": "working", "ttlMs": "PT2H", "pollIntervalMs": "PT10S" }""" - val res = decode[CreateTaskResult](json) - res.map(_.ttl) shouldBe Right(Some(2.hours)) - res.map(_.pollInterval) shouldBe Right(Some(10.seconds)) + json.hcursor.downField("ttlMs").as[Long] shouldBe Right(3600000L) + json.hcursor.downField("pollIntervalMs").as[Long] shouldBe Right(5000L) it should "encode and decode task status with the spec wire strings" in: (TaskStatus.InputRequired: TaskStatus).asJson shouldBe Json.fromString("input_required") From 01c2248aca010587f9257d545edcdb0cf8662f4e Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 00:44:53 +0200 Subject: [PATCH 09/15] feat: run tasks on virtual threads (JDK 21) Default TaskExecutor.threadPool to Executors.newVirtualThreadPerTaskExecutor() (JDK 21+), which fits the blocking tool logic tasks run and scales to many concurrent tasks. chimp's synchronous stack (ox, tapir-netty-server-sync) already requires JDK 21 at runtime. Co-Authored-By: Claude Opus 4.8 --- server/src/main/scala/chimp/server/TaskSupport.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/server/src/main/scala/chimp/server/TaskSupport.scala b/server/src/main/scala/chimp/server/TaskSupport.scala index 6d92491..b68ccd5 100644 --- a/server/src/main/scala/chimp/server/TaskSupport.scala +++ b/server/src/main/scala/chimp/server/TaskSupport.scala @@ -39,9 +39,11 @@ trait TaskExecutor[F[_]]: object TaskExecutor: - /** A thread-pool executor for synchronous (`Identity`) servers, such as the Netty sync server. Cancellation interrupts the worker thread. + /** A [[TaskExecutor]] for synchronous (`Identity`) servers, such as the Netty sync server, backed by an `ExecutorService`. Defaults to a + * virtual-thread-per-task executor (JDK 21+), which suits the blocking tool logic that tasks run and scales to many concurrent tasks. + * Cancellation interrupts the worker thread. */ - def threadPool(pool: ExecutorService = Executors.newCachedThreadPool()): TaskExecutor[Identity] = new TaskExecutor[Identity]: + def threadPool(pool: ExecutorService = Executors.newVirtualThreadPerTaskExecutor()): TaskExecutor[Identity] = new TaskExecutor[Identity]: private val running = ConcurrentHashMap[TaskId, JavaFuture[?]]() def start(taskId: TaskId, body: () => Identity[Unit]): Identity[Unit] = From 1d41e7c5d91697926a8cb803af7e7f3a26973707 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 00:45:42 +0200 Subject: [PATCH 10/15] ci: build the release/publish job on JDK 21 The task executor uses Executors.newVirtualThreadPerTaskExecutor() (JDK 21+), but the reusable publish workflow otherwise compiles on JDK 11 by default, so pin the publish job to java-version 21. Co-Authored-By: Claude Opus 4.8 --- .github/workflows/ci.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 614be8f..4de0e2f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -64,6 +64,7 @@ jobs: if: github.event_name != 'pull_request' && (startsWith(github.ref, 'refs/tags/v')) secrets: inherit with: + java-version: "21" java-opts: "-Xmx4G" sttp-native: 1 @@ -76,4 +77,4 @@ jobs: if: github.event.pull_request.user.login == 'softwaremill-ci' needs: [ build, label ] uses: softwaremill/github-actions-workflows/.github/workflows/auto-merge.yml@main - secrets: inherit \ No newline at end of file + secrets: inherit From 6250fd76549bc76823637a3f4cf0ad7d63b233f5 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 01:07:58 +0200 Subject: [PATCH 11/15] refactor: derive Codec for task results; name duration fields after wire keys Replace the hand-written Codec.forProductN for CreateTaskResult and GetTaskResult with `derives Codec`, naming the duration fields ttlMs / pollIntervalMs (FiniteDuration) so the derived keys match the spec wire without a manual codec that could drift. Co-Authored-By: Claude Opus 4.8 --- .../src/main/scala/chimp/protocol/Tasks.scala | 54 ++++--------------- .../test/scala/chimp/protocol/TasksSpec.scala | 8 +-- .../main/scala/chimp/server/McpHandler.scala | 8 +-- 3 files changed, 18 insertions(+), 52 deletions(-) diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index a262100..9a5305d 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -59,24 +59,21 @@ object TaskStatus: given Encoder[TaskStatus] = Encoder.instance(status => Json.fromString(toWire(status))) given Decoder[TaskStatus] = Decoder.decodeString.emap(s => fromWire.get(s).toRight(s"Unknown task status: $s")) -/** Result returned when a receiver answers a request with a task instead of the request's normal result. */ +/** Result returned when a receiver answers a request with a task instead of the request's normal result. `ttlMs` and `pollIntervalMs` carry + * their unit in the name because that is the wire field name; the values are typed as [[scala.concurrent.duration.FiniteDuration]] and + * serialized as integer milliseconds. + */ final case class CreateTaskResult( taskId: TaskId, status: TaskStatus, createdAt: Option[Instant] = None, lastUpdatedAt: Option[Instant] = None, - ttl: Option[FiniteDuration] = None, - pollInterval: Option[FiniteDuration] = None, + ttlMs: Option[FiniteDuration] = None, + pollIntervalMs: Option[FiniteDuration] = None, statusMessage: Option[String] = None, resultType: String = "task", _meta: Option[Map[String, Json]] = None -) - -object CreateTaskResult: - given Codec[CreateTaskResult] = - Codec.forProduct9("taskId", "status", "createdAt", "lastUpdatedAt", "ttlMs", "pollIntervalMs", "statusMessage", "resultType", "_meta")( - CreateTaskResult.apply - )(r => (r.taskId, r.status, r.createdAt, r.lastUpdatedAt, r.ttl, r.pollInterval, r.statusMessage, r.resultType, r._meta)) +) derives Codec final case class GetTaskParams(taskId: TaskId, _meta: Option[Map[String, Json]] = None) derives Codec final case class GetTaskRequest(method: String = "tasks/get", params: GetTaskParams) derives Codec @@ -89,46 +86,15 @@ final case class GetTaskResult( status: TaskStatus, createdAt: Option[Instant] = None, lastUpdatedAt: Option[Instant] = None, - ttl: Option[FiniteDuration] = None, - pollInterval: Option[FiniteDuration] = None, + ttlMs: Option[FiniteDuration] = None, + pollIntervalMs: Option[FiniteDuration] = None, statusMessage: Option[String] = None, result: Option[Json] = None, error: Option[JSONRPCErrorObject] = None, inputRequests: Option[Map[String, Json]] = None, resultType: Option[String] = None, _meta: Option[Map[String, Json]] = None -) - -object GetTaskResult: - given Codec[GetTaskResult] = Codec.forProduct12( - "taskId", - "status", - "createdAt", - "lastUpdatedAt", - "ttlMs", - "pollIntervalMs", - "statusMessage", - "result", - "error", - "inputRequests", - "resultType", - "_meta" - )(GetTaskResult.apply)(r => - ( - r.taskId, - r.status, - r.createdAt, - r.lastUpdatedAt, - r.ttl, - r.pollInterval, - r.statusMessage, - r.result, - r.error, - r.inputRequests, - r.resultType, - r._meta - ) - ) +) derives Codec final case class CancelTaskParams(taskId: TaskId, _meta: Option[Map[String, Json]] = None) derives Codec final case class CancelTaskRequest(method: String = "tasks/cancel", params: CancelTaskParams) derives Codec diff --git a/core/src/test/scala/chimp/protocol/TasksSpec.scala b/core/src/test/scala/chimp/protocol/TasksSpec.scala index b0665b6..627186a 100644 --- a/core/src/test/scala/chimp/protocol/TasksSpec.scala +++ b/core/src/test/scala/chimp/protocol/TasksSpec.scala @@ -28,16 +28,16 @@ class TasksSpec extends AnyFlatSpec with Matchers: res.map(_.taskId) shouldBe Right(TaskId("786512e2-9e0d-44bd-8f29-789f320fe840")) res.map(_.status) shouldBe Right(TaskStatus.Working) res.map(_.createdAt) shouldBe Right(Some(Instant.parse("2025-11-25T10:30:00Z"))) - res.map(_.ttl) shouldBe Right(Some(1.hour)) - res.map(_.pollInterval) shouldBe Right(Some(5.seconds)) + res.map(_.ttlMs) shouldBe Right(Some(1.hour)) + res.map(_.pollIntervalMs) shouldBe Right(Some(5.seconds)) it should "encode durations as integer milliseconds on the wire" in: val created = CreateTaskResult( taskId = TaskId("t"), status = TaskStatus.Working, - ttl = Some(1.hour), - pollInterval = Some(5.seconds) + ttlMs = Some(1.hour), + pollIntervalMs = Some(5.seconds) ) val json = created.asJson json.hcursor.downField("ttlMs").as[Long] shouldBe Right(3600000L) diff --git a/server/src/main/scala/chimp/server/McpHandler.scala b/server/src/main/scala/chimp/server/McpHandler.scala index 593b383..d432bfa 100644 --- a/server/src/main/scala/chimp/server/McpHandler.scala +++ b/server/src/main/scala/chimp/server/McpHandler.scala @@ -208,8 +208,8 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD status = TaskStatus.Working, createdAt = Some(now), lastUpdatedAt = Some(now), - ttl = support.ttl, - pollInterval = support.pollInterval, + ttlMs = support.ttl, + pollIntervalMs = support.pollInterval, resultType = Some("complete") ) // handleError takes its body by-name, so a synchronous (Identity) tool that throws is caught here too @@ -237,8 +237,8 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD status = TaskStatus.Working, createdAt = Some(now), lastUpdatedAt = Some(now), - ttl = support.ttl, - pollInterval = support.pollInterval + ttlMs = support.ttl, + pollIntervalMs = support.pollInterval ).asJson ) From 9178a82e5bdc29cd429cc975df9faa51614c86b7 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 01:14:32 +0200 Subject: [PATCH 12/15] refactor: model detailed task state as a sealed TaskOutcome Replace the loose status / result / error / inputRequests optional fields on GetTaskResult with a sealed TaskOutcome (Working, InputRequired(inputRequests), Completed(result), Failed(error), Cancelled), mirroring the ext-tasks DetailedTask union. status is derived from the outcome, so result/error/ inputRequests can only exist with the matching status. A hand-written codec flattens the outcome onto the wire (status plus result/error/inputRequests) and rejects, for example, a completed task with no result. Co-Authored-By: Claude Opus 4.8 --- .../scala/chimp/client/TasksClientSpec.scala | 8 ++- .../src/main/scala/chimp/protocol/Tasks.scala | 70 +++++++++++++++++-- .../test/scala/chimp/protocol/TasksSpec.scala | 9 ++- .../main/scala/chimp/server/McpHandler.scala | 22 ++---- .../scala/chimp/server/TaskServerSpec.scala | 15 ++-- 5 files changed, 89 insertions(+), 35 deletions(-) diff --git a/client/src/test/scala/chimp/client/TasksClientSpec.scala b/client/src/test/scala/chimp/client/TasksClientSpec.scala index 04d8c4e..e6f1feb 100644 --- a/client/src/test/scala/chimp/client/TasksClientSpec.scala +++ b/client/src/test/scala/chimp/client/TasksClientSpec.scala @@ -34,8 +34,7 @@ class TasksClientSpec extends AnyFlatSpec with Matchers: it should "poll a task with tasks/get and expose the underlying result" in: val task = GetTaskResult( taskId = TaskId("t1"), - status = TaskStatus.Completed, - result = Some(CallToolResult(content = List(ToolContent.Text(text = "done"))).asJson), + outcome = TaskOutcome.Completed(CallToolResult(content = List(ToolContent.Text(text = "done"))).asJson), resultType = Some("complete") ) val taskEnvelope = (JSONRPCMessage.Response(id = RequestId(1), result = task.asJson): JSONRPCMessage).asJson.noSpaces @@ -49,7 +48,10 @@ class TasksClientSpec extends AnyFlatSpec with Matchers: val res = client(backend).getTask(TaskId("t1")) res.status shouldBe TaskStatus.Completed - res.result.flatMap(_.as[CallToolResult].toOption).map(_.content.head) shouldBe Some(ToolContent.Text("text", "done")) + res.outcome match + case TaskOutcome.Completed(result) => + result.as[CallToolResult].toOption.map(_.content.head) shouldBe Some(ToolContent.Text("text", "done")) + case other => fail(s"expected Completed, got $other") it should "cancel a task with tasks/cancel" in: val ack = TaskAck(taskId = Some(TaskId("t1")), status = Some(TaskStatus.Cancelled)) diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index 9a5305d..f57daab 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -1,6 +1,7 @@ package chimp.protocol import io.circe.{Codec, Decoder, Encoder, Json} +import io.circe.syntax.* import java.time.Instant import scala.concurrent.duration.{DurationLong, FiniteDuration} @@ -78,23 +79,78 @@ final case class CreateTaskResult( final case class GetTaskParams(taskId: TaskId, _meta: Option[Map[String, Json]] = None) derives Codec final case class GetTaskRequest(method: String = "tasks/get", params: GetTaskParams) derives Codec -/** Detailed task state returned by `tasks/get`. `result` is present once the task is `Completed`, `error` once it has `Failed`, and - * `inputRequests` while it is `InputRequired`. `result` is left as raw JSON, since its shape depends on the request the task stands for. +/** The status of a task together with the data specific to that status. Modelling it as a sealed type means `result`, `error` and + * `inputRequests` can only appear with the status they belong to. `result` is raw JSON, since its shape depends on the request the task + * stands for. + */ +enum TaskOutcome: + case Working + case InputRequired(inputRequests: Map[String, Json]) + case Completed(result: Json) + case Failed(error: JSONRPCErrorObject) + case Cancelled + + def status: TaskStatus = this match + case TaskOutcome.Working => TaskStatus.Working + case TaskOutcome.InputRequired(_) => TaskStatus.InputRequired + case TaskOutcome.Completed(_) => TaskStatus.Completed + case TaskOutcome.Failed(_) => TaskStatus.Failed + case TaskOutcome.Cancelled => TaskStatus.Cancelled + +/** Detailed task state returned by `tasks/get`. The [[TaskOutcome]] carries the `status` and its status-specific data. On the wire the + * outcome is flattened: `status` plus, where applicable, `result` / `error` / `inputRequests`. */ final case class GetTaskResult( taskId: TaskId, - status: TaskStatus, + outcome: TaskOutcome, createdAt: Option[Instant] = None, lastUpdatedAt: Option[Instant] = None, ttlMs: Option[FiniteDuration] = None, pollIntervalMs: Option[FiniteDuration] = None, statusMessage: Option[String] = None, - result: Option[Json] = None, - error: Option[JSONRPCErrorObject] = None, - inputRequests: Option[Map[String, Json]] = None, resultType: Option[String] = None, _meta: Option[Map[String, Json]] = None -) derives Codec +): + def status: TaskStatus = outcome.status + +object GetTaskResult: + given Encoder[GetTaskResult] = Encoder.instance: task => + val base = Json.obj( + "taskId" -> task.taskId.asJson, + "status" -> task.status.asJson, + "createdAt" -> task.createdAt.asJson, + "lastUpdatedAt" -> task.lastUpdatedAt.asJson, + "ttlMs" -> task.ttlMs.asJson, + "pollIntervalMs" -> task.pollIntervalMs.asJson, + "statusMessage" -> task.statusMessage.asJson, + "resultType" -> task.resultType.asJson, + "_meta" -> task._meta.asJson + ) + val payload = task.outcome match + case TaskOutcome.Completed(result) => Json.obj("result" -> result) + case TaskOutcome.Failed(error) => Json.obj("error" -> error.asJson) + case TaskOutcome.InputRequired(inputRequests) => Json.obj("inputRequests" -> inputRequests.asJson) + case TaskOutcome.Working | TaskOutcome.Cancelled => Json.obj() + base.deepMerge(payload) + + given Decoder[GetTaskResult] = Decoder.instance: c => + for + taskId <- c.get[TaskId]("taskId") + status <- c.get[TaskStatus]("status") + createdAt <- c.get[Option[Instant]]("createdAt") + lastUpdatedAt <- c.get[Option[Instant]]("lastUpdatedAt") + ttlMs <- c.get[Option[FiniteDuration]]("ttlMs") + pollIntervalMs <- c.get[Option[FiniteDuration]]("pollIntervalMs") + statusMessage <- c.get[Option[String]]("statusMessage") + resultType <- c.get[Option[String]]("resultType") + meta <- c.get[Option[Map[String, Json]]]("_meta") + outcome <- status match + case TaskStatus.Working => Right(TaskOutcome.Working) + case TaskStatus.Cancelled => Right(TaskOutcome.Cancelled) + case TaskStatus.Completed => c.get[Json]("result").map(result => TaskOutcome.Completed(result)) + case TaskStatus.Failed => c.get[JSONRPCErrorObject]("error").map(error => TaskOutcome.Failed(error)) + case TaskStatus.InputRequired => c.get[Map[String, Json]]("inputRequests").map(reqs => TaskOutcome.InputRequired(reqs)) + yield GetTaskResult(taskId, outcome, createdAt, lastUpdatedAt, ttlMs, pollIntervalMs, statusMessage, resultType, meta) final case class CancelTaskParams(taskId: TaskId, _meta: Option[Map[String, Json]] = None) derives Codec final case class CancelTaskRequest(method: String = "tasks/cancel", params: CancelTaskParams) derives Codec diff --git a/core/src/test/scala/chimp/protocol/TasksSpec.scala b/core/src/test/scala/chimp/protocol/TasksSpec.scala index 627186a..ca9bb56 100644 --- a/core/src/test/scala/chimp/protocol/TasksSpec.scala +++ b/core/src/test/scala/chimp/protocol/TasksSpec.scala @@ -55,12 +55,17 @@ class TasksSpec extends AnyFlatSpec with Matchers: val toolResult = CallToolResult(content = List(ToolContent.Text(text = "Hello, Luca!"))).asJson val task = GetTaskResult( taskId = TaskId("t1"), - status = TaskStatus.Completed, - result = Some(toolResult), + outcome = TaskOutcome.Completed(toolResult), resultType = Some("complete") ) decode[GetTaskResult](task.asJson.noSpaces) shouldBe Right(task) + it should "flatten the outcome onto the wire and reject a completed task without a result" in: + val completed = GetTaskResult(taskId = TaskId("t1"), outcome = TaskOutcome.Completed(Json.obj("k" -> Json.fromInt(1)))) + completed.asJson.hcursor.downField("status").as[String] shouldBe Right("completed") + completed.asJson.hcursor.downField("result").as[Json] shouldBe Right(Json.obj("k" -> Json.fromInt(1))) + decode[GetTaskResult]("""{ "taskId": "t1", "status": "completed" }""").isLeft shouldBe true + it should "mark only completed, failed and cancelled as terminal" in: TaskStatus.values.filter(TaskStatus.isTerminal).toSet shouldBe Set(TaskStatus.Completed, TaskStatus.Failed, TaskStatus.Cancelled) diff --git a/server/src/main/scala/chimp/server/McpHandler.scala b/server/src/main/scala/chimp/server/McpHandler.scala index d432bfa..2f5b61b 100644 --- a/server/src/main/scala/chimp/server/McpHandler.scala +++ b/server/src/main/scala/chimp/server/McpHandler.scala @@ -205,7 +205,7 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD val now = java.time.Instant.now() val initial = GetTaskResult( taskId = taskId, - status = TaskStatus.Working, + outcome = TaskOutcome.Working, createdAt = Some(now), lastUpdatedAt = Some(now), ttlMs = support.ttl, @@ -216,14 +216,13 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD val body: () => F[Unit] = () => m.handleError( m.flatMap(tool.logic(input, context, headers))(result => - finishTask(support, taskId, TaskStatus.Completed, result = Some(toCallToolResult(result).asJson)) + finishTask(support, taskId, TaskOutcome.Completed(toCallToolResult(result).asJson)) ) ) { case t => finishTask( support, taskId, - TaskStatus.Failed, - error = Some(JSONRPCErrorObject(JSONRPCErrorCodes.InternalError.code, Option(t.getMessage).getOrElse("Task failed"))) + TaskOutcome.Failed(JSONRPCErrorObject(JSONRPCErrorCodes.InternalError.code, Option(t.getMessage).getOrElse("Task failed"))) ) } support.store @@ -243,19 +242,10 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD ) // only transition a task that is still working, so a cancellation is not overwritten by a late completion - private def finishTask( - support: TaskSupport[F], - taskId: TaskId, - status: TaskStatus, - result: Option[Json] = None, - error: Option[JSONRPCErrorObject] = None - )(using - MonadError[F] - ): F[Unit] = + private def finishTask(support: TaskSupport[F], taskId: TaskId, outcome: TaskOutcome)(using MonadError[F]): F[Unit] = support.store .update(taskId): current => - if current.status == TaskStatus.Working then - current.copy(status = status, result = result, error = error, lastUpdatedAt = Some(java.time.Instant.now())) + if current.status == TaskStatus.Working then current.copy(outcome = outcome, lastUpdatedAt = Some(java.time.Instant.now())) else current .map(_ => ()) @@ -273,7 +263,7 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD support.store .update(p.taskId): current => if TaskStatus.isTerminal(current.status) then current - else current.copy(status = TaskStatus.Cancelled, lastUpdatedAt = Some(java.time.Instant.now())) + else current.copy(outcome = TaskOutcome.Cancelled, lastUpdatedAt = Some(java.time.Instant.now())) .flatMap: case Some(_) => support.executor.cancel(p.taskId).map(_ => taskAck(id, p.taskId, TaskStatus.Cancelled)) case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}").unit diff --git a/server/src/test/scala/chimp/server/TaskServerSpec.scala b/server/src/test/scala/chimp/server/TaskServerSpec.scala index f5753b9..bcb0b6a 100644 --- a/server/src/test/scala/chimp/server/TaskServerSpec.scala +++ b/server/src/test/scala/chimp/server/TaskServerSpec.scala @@ -60,7 +60,7 @@ class TaskServerSpec extends AnyFlatSpec with Matchers: (Request(method = "tools/call", params = Some(params), id = RequestId("call")): JSONRPCMessage).asJson private def pollTask(handler: McpHandler[Identity, ServerContext[Identity]], taskId: TaskId): GetTaskResult = - var last = GetTaskResult(taskId = taskId, status = TaskStatus.Working) + var last = GetTaskResult(taskId = taskId, outcome = TaskOutcome.Working) var done = false var i = 0 while !done && i < 200 do @@ -87,10 +87,10 @@ class TaskServerSpec extends AnyFlatSpec with Matchers: created.taskId.value should not be empty val finished = pollTask(handler, created.taskId) - finished.status shouldBe TaskStatus.Completed - finished.result - .flatMap(_.as[CallToolResult].toOption) - .map(_.content.head) shouldBe Some(ToolContent.Text("text", "slow:hi")) + finished.outcome match + case TaskOutcome.Completed(result) => + result.as[CallToolResult].toOption.map(_.content.head) shouldBe Some(ToolContent.Text("text", "slow:hi")) + case other => fail(s"expected Completed, got $other") it should "report a failed task when the tool throws" in: val handler = handlerWith() @@ -99,8 +99,9 @@ class TaskServerSpec extends AnyFlatSpec with Matchers: .getOrElse(fail("decode CreateTaskResult")) val finished = pollTask(handler, created.taskId) - finished.status shouldBe TaskStatus.Failed - finished.error.isDefined shouldBe true + finished.outcome match + case TaskOutcome.Failed(_) => succeed + case other => fail(s"expected Failed, got $other") it should "cancel a running task" in: val handler = handlerWith() From 1d9e276d8a2e349d4f60a9508f9024f33158883e Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 01:26:43 +0200 Subject: [PATCH 13/15] refactor: return a sealed ToolCallResponse from callToolWithTasks Replace Either[CallToolResult, CreateTaskResult] with a named sealed type ToolCallResponse (Immediate(result) / Deferred(task)), decoded from the resultType discriminator, so call sites match on meaningful cases instead of Right/Left. Co-Authored-By: Claude Opus 4.8 --- client/src/main/scala/chimp/client/McpClient.scala | 8 ++++---- .../main/scala/chimp/client/McpClientImpl.scala | 13 ++----------- .../test/scala/chimp/client/TasksClientSpec.scala | 2 +- core/src/main/scala/chimp/protocol/Tasks.scala | 14 ++++++++++++++ 4 files changed, 21 insertions(+), 16 deletions(-) diff --git a/client/src/main/scala/chimp/client/McpClient.scala b/client/src/main/scala/chimp/client/McpClient.scala index 60b4d1f..24307ff 100644 --- a/client/src/main/scala/chimp/client/McpClient.scala +++ b/client/src/main/scala/chimp/client/McpClient.scala @@ -106,11 +106,11 @@ trait McpClient[F[_]]: /** Fulfils the input a task is waiting for while it is `InputRequired` (MCP Tasks extension, experimental). */ def updateTask(taskId: TaskId, inputResponses: Map[String, Json]): F[Unit] - /** Invokes a tool, declaring support for the Tasks extension (experimental). The server may answer either directly with a - * [[chimp.protocol.CallToolResult]] (`Left`) or, for a long-running call, with a [[chimp.protocol.CreateTaskResult]] task handle - * (`Right`) that is then driven with [[getTask]] / [[cancelTask]] / [[updateTask]]. + /** Invokes a tool, declaring support for the Tasks extension (experimental). The server may answer directly + * ([[chimp.protocol.ToolCallResponse.Immediate]]) or, for a long-running call, defer with a task handle + * ([[chimp.protocol.ToolCallResponse.Deferred]]) that is then driven with [[getTask]] / [[cancelTask]] / [[updateTask]]. */ - def callToolWithTasks(name: String, arguments: Json): F[Either[CallToolResult, CreateTaskResult]] + def callToolWithTasks(name: String, arguments: Json): F[ToolCallResponse] /** An [[McpClient]] used over a [[chimp.client.transport.ClientBidirectionalTransport]], which additionally supports server-initiated * interactions: subscribing to resource updates, notifying the server about changes to the client's roots, and handling notifications diff --git a/client/src/main/scala/chimp/client/McpClientImpl.scala b/client/src/main/scala/chimp/client/McpClientImpl.scala index d52a3fa..e2060a5 100644 --- a/client/src/main/scala/chimp/client/McpClientImpl.scala +++ b/client/src/main/scala/chimp/client/McpClientImpl.scala @@ -229,19 +229,10 @@ object McpClientImpl: override def updateTask(taskId: TaskId, inputResponses: Map[String, Json]): F[Unit] = sendRequest[Json]("tasks/update", Some(UpdateTaskParams(taskId, inputResponses).asJson)).map(_ => ()) - override def callToolWithTasks(name: String, arguments: Json): F[Either[CallToolResult, CreateTaskResult]] = + override def callToolWithTasks(name: String, arguments: Json): F[ToolCallResponse] = requireServerCapability("tools/call", _.tools.isDefined): val params = CallToolParams(name = name, arguments = arguments, _meta = Some(TasksExtension.clientCapabilityMeta)).asJson - sendRequest[Json]("tools/call", Some(params)).flatMap: json => - val isTask = json.hcursor.downField("resultType").as[String].toOption.contains("task") - if isTask then - json.as[CreateTaskResult] match - case Right(task) => monad.unit(Right(task)) - case Left(error) => monad.error(McpProtocolException(s"Failed to decode CreateTaskResult: ${error.getMessage}")) - else - json.as[CallToolResult] match - case Right(result) => monad.unit(Left(result)) - case Left(error) => monad.error(McpProtocolException(s"Failed to decode CallToolResult: ${error.getMessage}")) + sendRequest[ToolCallResponse]("tools/call", Some(params)) protected def requireServerCapability[A](method: String, present: ServerCapabilities => Boolean)(action: => F[A]): F[A] = if present(serverCapabilities) then action diff --git a/client/src/test/scala/chimp/client/TasksClientSpec.scala b/client/src/test/scala/chimp/client/TasksClientSpec.scala index e6f1feb..b2c71ab 100644 --- a/client/src/test/scala/chimp/client/TasksClientSpec.scala +++ b/client/src/test/scala/chimp/client/TasksClientSpec.scala @@ -83,4 +83,4 @@ class TasksClientSpec extends AnyFlatSpec with Matchers: .whenAnyRequest .thenRespondAdjust("", StatusCode.Accepted) - client(backend).callToolWithTasks("slow", io.circe.Json.obj()) shouldBe Right(created) + client(backend).callToolWithTasks("slow", io.circe.Json.obj()) shouldBe ToolCallResponse.Deferred(created) diff --git a/core/src/main/scala/chimp/protocol/Tasks.scala b/core/src/main/scala/chimp/protocol/Tasks.scala index f57daab..928bd82 100644 --- a/core/src/main/scala/chimp/protocol/Tasks.scala +++ b/core/src/main/scala/chimp/protocol/Tasks.scala @@ -76,6 +76,20 @@ final case class CreateTaskResult( _meta: Option[Map[String, Json]] = None ) derives Codec +/** The response to a `tools/call` made with task support declared: the receiver either answers immediately with the tool's + * [[CallToolResult]], or defers by returning a [[CreateTaskResult]] task handle to poll. + */ +enum ToolCallResponse: + case Immediate(result: CallToolResult) + case Deferred(task: CreateTaskResult) + +object ToolCallResponse: + given Decoder[ToolCallResponse] = Decoder.instance: c => + c.get[Option[String]]("resultType") + .flatMap: + case Some("task") => c.as[CreateTaskResult].map(ToolCallResponse.Deferred(_)) + case _ => c.as[CallToolResult].map(ToolCallResponse.Immediate(_)) + final case class GetTaskParams(taskId: TaskId, _meta: Option[Map[String, Json]] = None) derives Codec final case class GetTaskRequest(method: String = "tasks/get", params: GetTaskParams) derives Codec From 2f2a96448b8dfc26027a5e256e3e54a46c31ab23 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 01:26:43 +0200 Subject: [PATCH 14/15] refactor: pass the task body to TaskExecutor.start by name Change start(taskId, body: () => F[Unit]) to start(taskId)(body: => F[Unit]), so callers pass the task body as a by-name argument rather than an explicit thunk. Deferral of a synchronous (Identity) body to the worker thread is preserved. Co-Authored-By: Claude Opus 4.8 --- .../main/scala/chimp/server/McpHandler.scala | 33 ++++++++++--------- .../main/scala/chimp/server/TaskSupport.scala | 6 ++-- 2 files changed, 21 insertions(+), 18 deletions(-) diff --git a/server/src/main/scala/chimp/server/McpHandler.scala b/server/src/main/scala/chimp/server/McpHandler.scala index 2f5b61b..4f3977a 100644 --- a/server/src/main/scala/chimp/server/McpHandler.scala +++ b/server/src/main/scala/chimp/server/McpHandler.scala @@ -212,23 +212,25 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD pollIntervalMs = support.pollInterval, resultType = Some("complete") ) - // handleError takes its body by-name, so a synchronous (Identity) tool that throws is caught here too - val body: () => F[Unit] = () => - m.handleError( - m.flatMap(tool.logic(input, context, headers))(result => - finishTask(support, taskId, TaskOutcome.Completed(toCallToolResult(result).asJson)) - ) - ) { case t => - finishTask( - support, - taskId, - TaskOutcome.Failed(JSONRPCErrorObject(JSONRPCErrorCodes.InternalError.code, Option(t.getMessage).getOrElse("Task failed"))) - ) - } + // start and handleError both take their body by-name, so a synchronous (Identity) tool that throws is caught here too support.store .create(initial) - .flatMap(_ => support.executor.start(taskId, body)) - .map: _ => + .flatMap { _ => + support.executor.start(taskId)( + m.handleError( + m.flatMap(tool.logic(input, context, headers))(result => + finishTask(support, taskId, TaskOutcome.Completed(toCallToolResult(result).asJson)) + ) + ) { case t => + finishTask( + support, + taskId, + TaskOutcome.Failed(JSONRPCErrorObject(JSONRPCErrorCodes.InternalError.code, Option(t.getMessage).getOrElse("Task failed"))) + ) + } + ) + } + .map { _ => JSONRPCMessage.Response( id = id, result = CreateTaskResult( @@ -240,6 +242,7 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD pollIntervalMs = support.pollInterval ).asJson ) + } // only transition a task that is still working, so a cancellation is not overwritten by a late completion private def finishTask(support: TaskSupport[F], taskId: TaskId, outcome: TaskOutcome)(using MonadError[F]): F[Unit] = diff --git a/server/src/main/scala/chimp/server/TaskSupport.scala b/server/src/main/scala/chimp/server/TaskSupport.scala index b68ccd5..0db26ea 100644 --- a/server/src/main/scala/chimp/server/TaskSupport.scala +++ b/server/src/main/scala/chimp/server/TaskSupport.scala @@ -34,7 +34,7 @@ object TaskStore: * such as `Identity`, it is only run on the background worker rather than at the call site. */ trait TaskExecutor[F[_]]: - def start(taskId: TaskId, body: () => F[Unit]): F[Unit] + def start(taskId: TaskId)(body: => F[Unit]): F[Unit] def cancel(taskId: TaskId): F[Unit] object TaskExecutor: @@ -46,10 +46,10 @@ object TaskExecutor: def threadPool(pool: ExecutorService = Executors.newVirtualThreadPerTaskExecutor()): TaskExecutor[Identity] = new TaskExecutor[Identity]: private val running = ConcurrentHashMap[TaskId, JavaFuture[?]]() - def start(taskId: TaskId, body: () => Identity[Unit]): Identity[Unit] = + def start(taskId: TaskId)(body: => Identity[Unit]): Identity[Unit] = val future = pool.submit(new Runnable: def run(): Unit = - try body() + try body finally val _ = running.remove(taskId)) val _ = running.put(taskId, future) From 8a997389d8f44b043f41fdbd87ef4e3e849f04ba Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Tue, 1 Sep 2026 01:53:23 +0200 Subject: [PATCH 15/15] feat: implement server-initiated input_required task flows A tool registered with addTaskTool (defined via Tool.taskLogic) receives a TaskContext and can call requestInput(key, request) while running as a task. This moves the task to input_required, surfacing key -> request via tasks/get; the tool parks (on the virtual-thread executor) until the client answers with tasks/update, then resumes, transitioning back to working (or completing once all inputs are resolved). - TaskContext capability + Tool.taskLogic builder + McpServer/StreamingMcpServer addTaskTool and a taskTools list. - TaskInputCoordinator parks the worker and delivers answers; the waiter is registered before input_required is advertised, so a fast tasks/update is never lost. Cancelling a task unblocks its waiters. - tasks/update now delivers inputResponses; task tools are always answered with a task and require the client capability (-32003 otherwise). Task tools also appear in tools/list. Co-Authored-By: Claude Opus 4.8 --- docs/server/capabilities.md | 30 +++- .../main/scala/chimp/server/McpHandler.scala | 132 ++++++++++++------ .../main/scala/chimp/server/McpServer.scala | 18 ++- .../scala/chimp/server/ServerContext.scala | 7 + .../main/scala/chimp/server/TaskSupport.scala | 31 +++- server/src/main/scala/chimp/server/Tool.scala | 6 + .../scala/chimp/server/TaskServerSpec.scala | 50 ++++++- 7 files changed, 224 insertions(+), 50 deletions(-) diff --git a/docs/server/capabilities.md b/docs/server/capabilities.md index f2b8280..d3df9f4 100644 --- a/docs/server/capabilities.md +++ b/docs/server/capabilities.md @@ -53,4 +53,32 @@ val server = McpServer(tools = List(report)) .withTasks(TaskSupport(TaskStore.inMemory[Identity], TaskExecutor.threadPool())) ``` -The server runs the tool in the background, transitions the task to `completed` (with the tool's `CallToolResult`) or `failed`, and answers `tasks/cancel` by interrupting the worker. `useTask` and `requireTask` on `TaskSupport` control, per tool, whether a task is offered or required; a required task with no client support fails with `-32003`. Full server-initiated `input_required` flows are not yet implemented. +The server runs the tool in the background, transitions the task to `completed` (with the tool's `CallToolResult`) or `failed`, and answers `tasks/cancel` by interrupting the worker. `useTask` and `requireTask` on `TaskSupport` control, per tool, whether a task is offered or required; a required task with no client support fails with `-32003`. + +### Requesting input while running + +A tool registered with `addTaskTool` (defined with `.taskLogic`) receives a `TaskContext` and can ask the client for input mid-run. `requestInput` moves the task to `input_required`, surfacing the request under a key via `tasks/get`; the tool parks until the client answers with `tasks/update`, then resumes: + +```scala mdoc:compile-only +import chimp.server.* +import io.circe.{Codec, Json} +import sttp.monad.{IdentityMonad, MonadError} +import sttp.shared.Identity +import sttp.tapir.Schema + +given MonadError[Identity] = IdentityMonad + +case class ReviewInput(text: String) derives Codec, Schema + +val review = tool("review") + .input[ReviewInput] + .taskLogic[Identity]: (in, ctx, _) => + val decision = ctx.requestInput("approve", Json.fromString(s"Approve: ${in.text}?")) + ToolResult.text(s"decision: ${decision.noSpaces}") + +val server = McpServer() + .withTasks(TaskSupport(TaskStore.inMemory[Identity], TaskExecutor.threadPool())) + .addTaskTool(review) +``` + +A task tool always runs as a task and requires the client to declare task support, so calling it without that support fails with `-32003`. diff --git a/server/src/main/scala/chimp/server/McpHandler.scala b/server/src/main/scala/chimp/server/McpHandler.scala index 4f3977a..6f59274 100644 --- a/server/src/main/scala/chimp/server/McpHandler.scala +++ b/server/src/main/scala/chimp/server/McpHandler.scala @@ -29,10 +29,12 @@ enum McpResponse: private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerDef[F, C]): private val logger = LoggerFactory.getLogger(classOf[McpHandler[?, ?]]) private val toolsByName = server.tools.map(tool => tool.name -> tool).toMap + private val taskToolsByName = server.taskTools.map(tool => tool.name -> tool).toMap + private val inputCoordinator = TaskInputCoordinator() private val promptsByName = server.prompts.map(prompt => prompt.definition.name -> prompt).toMap private val resourcesByUri = server.resources.map(resource => resource.definition.uri -> resource).toMap private val hasResources = server.resources.nonEmpty || server.resourceTemplates.nonEmpty - private val toolDefinitions = server.tools.map(toolToDefinition) + private val toolDefinitions = server.tools.map(toolToDefinition) ++ server.taskTools.map(toolToDefinition) private def toJsonSchema(toolSchema: ToolSchema): Json = toolSchema match case ToolSchema.Derived(schema) => @@ -40,7 +42,7 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD (if server.showJsonSchemaMetadata then base else base.copy($schema = None)).asJson case ToolSchema.Raw(json) => json - private def toolToDefinition(tool: ServerTool[?, ?, F, C]): ToolDefinition = + private def toolToDefinition(tool: ServerTool[?, ?, F, ?]): ToolDefinition = ToolDefinition( name = tool.name, description = tool.description, @@ -144,42 +146,59 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD JSONRPCMessage.Response(id = id, result = result.asJson) private def handleToolsCall(params: Option[Json], id: RequestId, headers: Seq[Header], makeContext: Option[ProgressToken] => C)(using - MonadError[F] + m: MonadError[F] ): F[JSONRPCMessage] = val name = params.flatMap(_.hcursor.downField("name").as[String].toOption) val arguments = params.flatMap(_.hcursor.downField("arguments").focus).getOrElse(Json.obj()) val progressToken = params.flatMap(_.hcursor.downField("_meta").downField("progressToken").as[ProgressToken].toOption) val requestMeta = params.flatMap(_.hcursor.downField("_meta").as[Map[String, Json]].toOption) val clientSupportsTasks = TasksExtension.declaredIn(requestMeta) + + def invalidArguments(error: DecodingFailure): F[JSONRPCMessage] = + protocolError( + id, + JSONRPCErrorCodes.InvalidParams.code, + s"Invalid arguments: ${error.getMessage}. Input: ${arguments.noSpaces.take(200)}" + ).unit + + def missingTaskCapability(toolName: String): F[JSONRPCMessage] = + protocolError( + id, + JSONRPCErrorCodes.MissingRequiredClientCapability.code, + s"Tool '$toolName' requires the ${TasksExtension.Id} client capability" + ).unit + name match case Some(name) => - toolsByName.get(name) match - case Some(tool) => - tool.inputDecoder.decodeJson(arguments) match - case Right(input) => - val context = makeContext(progressToken) - server.tasks match - case Some(support) if support.requireTask(name) && !clientSupportsTasks => - protocolError( - id, - JSONRPCErrorCodes.MissingRequiredClientCapability.code, - s"Tool '$name' requires the ${TasksExtension.Id} client capability" - ).unit - case Some(support) if clientSupportsTasks && support.useTask(name) => - startTask(support, id, tool, input, context, headers) - case _ => - tool - .logic(input, context, headers) - .map: result => - toolCallResponse(id, result) - case Left(decodingError) => - val snippet = arguments.noSpaces.take(200) + taskToolsByName.get(name) match + // a task tool always runs as a task and needs both server task support and the client capability + case Some(taskTool) => + server.tasks match + case None => protocolError( id, - JSONRPCErrorCodes.InvalidParams.code, - s"Invalid arguments: ${decodingError.getMessage}. Input: $snippet" + JSONRPCErrorCodes.InternalError.code, + s"Tool '$name' runs as a task, but task support is not configured" ).unit - case None => protocolError(id, JSONRPCErrorCodes.MethodNotFound.code, s"Unknown tool: $name").unit + case Some(_) if !clientSupportsTasks => missingTaskCapability(name) + case Some(support) => + taskTool.inputDecoder.decodeJson(arguments) match + case Right(input) => runTask(support, id)(taskId => taskTool.logic(input, makeTaskContext(support, taskId), headers)) + case Left(error) => invalidArguments(error) + case None => + toolsByName.get(name) match + case Some(tool) => + tool.inputDecoder.decodeJson(arguments) match + case Right(input) => + val context = makeContext(progressToken) + server.tasks match + case Some(support) if support.requireTask(name) && !clientSupportsTasks => missingTaskCapability(name) + case Some(support) if clientSupportsTasks && support.useTask(name) => + runTask(support, id)(_ => tool.logic(input, context, headers)) + case _ => + tool.logic(input, context, headers).map(result => toolCallResponse(id, result)) + case Left(error) => invalidArguments(error) + case None => protocolError(id, JSONRPCErrorCodes.MethodNotFound.code, s"Unknown tool: $name").unit case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, "Missing tool name").unit @@ -193,14 +212,9 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD private def toolCallResponse(id: RequestId, result: ToolResult[?]): JSONRPCMessage = JSONRPCMessage.Response(id = id, result = toCallToolResult(result).asJson) - private def startTask[I]( - support: TaskSupport[F], - id: RequestId, - tool: ServerTool[I, ?, F, C], - input: I, - context: C, - headers: Seq[Header] - )(using m: MonadError[F]): F[JSONRPCMessage] = + private def runTask[O](support: TaskSupport[F], id: RequestId)(compute: TaskId => F[ToolResult[O]])(using + m: MonadError[F] + ): F[JSONRPCMessage] = val taskId = TaskId(java.util.UUID.randomUUID().toString) val now = java.time.Instant.now() val initial = GetTaskResult( @@ -218,9 +232,7 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD .flatMap { _ => support.executor.start(taskId)( m.handleError( - m.flatMap(tool.logic(input, context, headers))(result => - finishTask(support, taskId, TaskOutcome.Completed(toCallToolResult(result).asJson)) - ) + m.flatMap(compute(taskId))(result => finishTask(support, taskId, TaskOutcome.Completed(toCallToolResult(result).asJson))) ) { case t => finishTask( support, @@ -244,6 +256,37 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD ) } + private def makeTaskContext(support: TaskSupport[F], taskId: TaskId)(using m: MonadError[F]): TaskContext[F] = + new TaskContext[F]: + def requestInput(key: String, request: Json): F[Json] = + // register the waiter before advertising input_required, so a fast tasks/update is never lost + m.flatMap(m.eval(inputCoordinator.register(taskId, key))) { waiter => + m.flatMap(setInputRequired(support, taskId, key, request)) { _ => + // waiter.get blocks the worker until tasks/update delivers the answer; fine on the virtual-thread executor + m.flatMap(m.eval(waiter.get()))(response => m.map(resolveInput(support, taskId, key))(_ => response)) + } + } + + private def setInputRequired(support: TaskSupport[F], taskId: TaskId, key: String, request: Json)(using MonadError[F]): F[Unit] = + support.store + .update(taskId): current => + val outstanding = current.outcome match + case TaskOutcome.InputRequired(requests) => requests + case _ => Map.empty[String, Json] + current.copy(outcome = TaskOutcome.InputRequired(outstanding + (key -> request)), lastUpdatedAt = Some(java.time.Instant.now())) + .map(_ => ()) + + private def resolveInput(support: TaskSupport[F], taskId: TaskId, key: String)(using MonadError[F]): F[Unit] = + support.store + .update(taskId): current => + current.outcome match + case TaskOutcome.InputRequired(requests) => + val remaining = requests - key + val outcome = if remaining.isEmpty then TaskOutcome.Working else TaskOutcome.InputRequired(remaining) + current.copy(outcome = outcome, lastUpdatedAt = Some(java.time.Instant.now())) + case _ => current + .map(_ => ()) + // only transition a task that is still working, so a cancellation is not overwritten by a late completion private def finishTask(support: TaskSupport[F], taskId: TaskId, outcome: TaskOutcome)(using MonadError[F]): F[Unit] = support.store @@ -268,16 +311,21 @@ private[server] class McpHandler[F[_], C <: ServerContext[F]](server: McpServerD if TaskStatus.isTerminal(current.status) then current else current.copy(outcome = TaskOutcome.Cancelled, lastUpdatedAt = Some(java.time.Instant.now())) .flatMap: - case Some(_) => support.executor.cancel(p.taskId).map(_ => taskAck(id, p.taskId, TaskStatus.Cancelled)) - case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}").unit + case Some(_) => + inputCoordinator.cancel(p.taskId) + support.executor.cancel(p.taskId).map(_ => taskAck(id, p.taskId, TaskStatus.Cancelled)) + case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}").unit private def handleTasksUpdate(params: Option[Json], id: RequestId)(using MonadError[F]): F[JSONRPCMessage] = decodeParams[UpdateTaskParams](params, id): p => server.tasks.get.store .get(p.taskId) .map: - case Some(task) => taskAck(id, task.taskId, task.status) - case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}") + case Some(task) => + // hand each response to the waiting task tool; it transitions the task back to working itself + p.inputResponses.foreach((key, response) => inputCoordinator.deliverInput(p.taskId, key, response)) + taskAck(id, task.taskId, task.status) + case None => protocolError(id, JSONRPCErrorCodes.InvalidParams.code, s"Unknown task: ${p.taskId}") private def taskAck(id: RequestId, taskId: TaskId, status: TaskStatus): JSONRPCMessage = JSONRPCMessage.Response(id = id, result = TaskAck(taskId = Some(taskId), status = Some(status)).asJson) diff --git a/server/src/main/scala/chimp/server/McpServer.scala b/server/src/main/scala/chimp/server/McpServer.scala index f8eb9a2..71232e4 100644 --- a/server/src/main/scala/chimp/server/McpServer.scala +++ b/server/src/main/scala/chimp/server/McpServer.scala @@ -27,6 +27,7 @@ sealed trait McpServerDef[F[_], C <: ServerContext[F]]: def loggingLevel: Option[SetLoggingLevelHandler[F]] def subscriptions: Option[ResourceSubscriptions[F]] def tasks: Option[TaskSupport[F]] + def taskTools: List[ServerTool[?, ?, F, TaskContext[F]]] case class McpServer[F[_]]( name: String = "Chimp MCP server", @@ -41,7 +42,8 @@ case class McpServer[F[_]]( completion: Option[CompletionHandler[F]] = None, loggingLevel: Option[SetLoggingLevelHandler[F]] = None, subscriptions: Option[ResourceSubscriptions[F]] = None, - tasks: Option[TaskSupport[F]] = None + tasks: Option[TaskSupport[F]] = None, + taskTools: List[ServerTool[?, ?, F, TaskContext[F]]] = Nil ) extends McpServerDef[F, ServerContext[F]]: def name(value: String): McpServer[F] = copy(name = value) @@ -94,6 +96,10 @@ case class McpServer[F[_]]( def withTasks(support: TaskSupport[F]): McpServer[F] = copy(tasks = Some(support)) + /** Registers a tool that runs as a task and can request input from the client while running (Tasks extension). */ + def addTaskTool(tool: ServerTool[?, ?, F, TaskContext[F]]): McpServer[F] = + copy(taskTools = taskTools :+ tool) + def endpoint(path: List[String]): ServerEndpoint[Any, F] = ServerHttpTransport(path).serve(this) def streaming: StreamingMcpServer[F] = @@ -110,7 +116,8 @@ case class McpServer[F[_]]( completion, loggingLevel, subscriptions, - tasks + tasks, + taskTools ) case class StreamingMcpServer[F[_]]( @@ -126,7 +133,8 @@ case class StreamingMcpServer[F[_]]( completion: Option[CompletionHandler[F]] = None, loggingLevel: Option[SetLoggingLevelHandler[F]] = None, subscriptions: Option[ResourceSubscriptions[F]] = None, - tasks: Option[TaskSupport[F]] = None + tasks: Option[TaskSupport[F]] = None, + taskTools: List[ServerTool[?, ?, F, TaskContext[F]]] = Nil ) extends McpServerDef[F, StreamingServerContext[F]]: def name(value: String): StreamingMcpServer[F] = copy(name = value) @@ -184,3 +192,7 @@ case class StreamingMcpServer[F[_]]( def withTasks(support: TaskSupport[F]): StreamingMcpServer[F] = copy(tasks = Some(support)) + + /** Registers a tool that runs as a task and can request input from the client while running (Tasks extension). */ + def addTaskTool(tool: ServerTool[?, ?, F, TaskContext[F]]): StreamingMcpServer[F] = + copy(taskTools = taskTools :+ tool) diff --git a/server/src/main/scala/chimp/server/ServerContext.scala b/server/src/main/scala/chimp/server/ServerContext.scala index c764818..e8cc027 100644 --- a/server/src/main/scala/chimp/server/ServerContext.scala +++ b/server/src/main/scala/chimp/server/ServerContext.scala @@ -10,6 +10,13 @@ trait ServerContext[F[_]] object ServerContext: def noop[F[_]]: ServerContext[F] = new ServerContext[F] {} +/** A [[ServerContext]] for a tool running as a task, letting it request input from the client mid-execution (Tasks extension, + * experimental). Requesting input moves the task to `input_required`, surfacing `key -> request` via `tasks/get`; the tool resumes with + * the response once the client answers with `tasks/update`. + */ +trait TaskContext[F[_]] extends ServerContext[F]: + def requestInput(key: String, request: Json): F[Json] + trait StreamingServerContext[F[_]] extends ServerContext[F]: def reportProgress(progress: Double, total: Option[Double] = None, message: Option[String] = None): F[Unit] def log(level: LoggingLevel, data: Json, logger: Option[String] = None): F[Unit] diff --git a/server/src/main/scala/chimp/server/TaskSupport.scala b/server/src/main/scala/chimp/server/TaskSupport.scala index 0db26ea..d6219db 100644 --- a/server/src/main/scala/chimp/server/TaskSupport.scala +++ b/server/src/main/scala/chimp/server/TaskSupport.scala @@ -1,10 +1,11 @@ package chimp.server import chimp.protocol.{GetTaskResult, TaskId} +import io.circe.Json import sttp.monad.MonadError import sttp.shared.Identity -import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Executors, Future as JavaFuture} +import java.util.concurrent.{CompletableFuture, ConcurrentHashMap, ExecutorService, Executors, Future as JavaFuture} import scala.concurrent.duration.{DurationInt, FiniteDuration} /** Durable-ish store of task state for the Tasks extension, addressable by task id. The default in-memory implementation keeps tasks for @@ -75,3 +76,31 @@ final case class TaskSupport[F[_]]( useTask: String => Boolean = (_: String) => true, requireTask: String => Boolean = (_: String) => false ) + +/** Coordinates the `input_required` flow within a process: a running task tool parks on the future from [[register]] for a `(taskId, key)` + * while `tasks/update` hands the answer over with [[deliverInput]]. Cancelling a task unblocks any of its waiters. + */ +private[server] final class TaskInputCoordinator: + private val pending = ConcurrentHashMap[(TaskId, String), CompletableFuture[Json]]() + + /** Registers interest in an answer for `(taskId, key)` and returns the future to block on. Call this before advertising the request + * (moving the task to `input_required`), so a fast `tasks/update` is never delivered before there is a waiter. + */ + def register(taskId: TaskId, key: String): CompletableFuture[Json] = + pending.computeIfAbsent((taskId, key), _ => CompletableFuture[Json]()) + + /** Hands `response` to a waiter, if any. Returns whether a waiter was present. */ + def deliverInput(taskId: TaskId, key: String, response: Json): Boolean = + Option(pending.remove((taskId, key))).exists { future => + future.complete(response) + true + } + + /** Unblocks all waiters for a cancelled task by failing their futures. */ + def cancel(taskId: TaskId): Unit = + val iterator = pending.entrySet().iterator() + while iterator.hasNext do + val entry = iterator.next() + if entry.getKey._1 == taskId then + entry.getValue.completeExceptionally(InterruptedException("task cancelled")) + iterator.remove() diff --git a/server/src/main/scala/chimp/server/Tool.scala b/server/src/main/scala/chimp/server/Tool.scala index 1e4f489..f09c28d 100644 --- a/server/src/main/scala/chimp/server/Tool.scala +++ b/server/src/main/scala/chimp/server/Tool.scala @@ -109,6 +109,12 @@ case class Tool[I, O]( ): ServerTool[I, O, F, StreamingServerContext[F]] = ServerTool(name, description, inputSchema, inputDecoder, outputSchema, annotations, logic) + /** Attaches effectful logic with access to a [[TaskContext]], so the tool can request input from the client while running as a task + * (Tasks extension). Register it with `addTaskTool`; such a tool is always answered with a task. + */ + def taskLogic[F[_]](logic: (I, TaskContext[F], Seq[Header]) => F[ToolResult[O]]): ServerTool[I, O, F, TaskContext[F]] = + ServerTool(name, description, inputSchema, inputDecoder, outputSchema, annotations, logic) + /** Attaches synchronous logic that also receives the request headers. */ def handleWithHeaders(logic: (I, Seq[Header]) => ToolResult[O]): ServerTool[I, O, Identity, ServerContext[Identity]] = ServerTool(name, description, inputSchema, inputDecoder, outputSchema, annotations, (i, _, headers) => logic(i, headers)) diff --git a/server/src/test/scala/chimp/server/TaskServerSpec.scala b/server/src/test/scala/chimp/server/TaskServerSpec.scala index bcb0b6a..7a34762 100644 --- a/server/src/test/scala/chimp/server/TaskServerSpec.scala +++ b/server/src/test/scala/chimp/server/TaskServerSpec.scala @@ -30,13 +30,19 @@ class TaskServerSpec extends AnyFlatSpec with Matchers: Thread.sleep(10000) ToolResult.text("done") + private val askThenEcho = tool("askThenEcho") + .input[TIn] + .taskLogic[Identity]: (in, ctx, _) => + val answer = ctx.requestInput("q1", Json.fromString(s"need input for ${in.message}")) + ToolResult.text(s"got:${answer.noSpaces}") + private def handlerWith(requireTask: String => Boolean = _ => false): McpHandler[Identity, ServerContext[Identity]] = val support = TaskSupport[Identity]( store = TaskStore.inMemory[Identity], executor = TaskExecutor.threadPool(), requireTask = requireTask ) - McpHandler(McpServer(tools = List(instant, slow, boom, forever)).withTasks(support)) + McpHandler(McpServer(tools = List(instant, slow, boom, forever)).withTasks(support).addTaskTool(askThenEcho)) private def resultJson(response: McpResponse): Json = val json = response match @@ -59,19 +65,24 @@ class TaskServerSpec extends AnyFlatSpec with Matchers: val params = CallToolParams(name = name, arguments = TIn("hi").asJson, _meta = meta).asJson (Request(method = "tools/call", params = Some(params), id = RequestId("call")): JSONRPCMessage).asJson - private def pollTask(handler: McpHandler[Identity, ServerContext[Identity]], taskId: TaskId): GetTaskResult = + private def pollUntil(handler: McpHandler[Identity, ServerContext[Identity]], taskId: TaskId)( + predicate: GetTaskResult => Boolean + ): GetTaskResult = var last = GetTaskResult(taskId = taskId, outcome = TaskOutcome.Working) var done = false var i = 0 while !done && i < 200 do val req = (Request(method = "tasks/get", params = Some(GetTaskParams(taskId).asJson), id = RequestId("get")): JSONRPCMessage).asJson last = resultJson(handler.handleJsonRpc(req, Seq.empty)).as[GetTaskResult].getOrElse(fail("decode GetTaskResult")) - if TaskStatus.isTerminal(last.status) then done = true + if predicate(last) then done = true else Thread.sleep(20) i += 1 last + private def pollTask(handler: McpHandler[Identity, ServerContext[Identity]], taskId: TaskId): GetTaskResult = + pollUntil(handler, taskId)(task => TaskStatus.isTerminal(task.status)) + "a task-enabled server" should "run tools/call synchronously when the client does not declare task support" in: val handler = handlerWith() val result = resultJson(handler.handleJsonRpc(callToolReq("instant", withTasks = false), Seq.empty)) @@ -126,3 +137,36 @@ class TaskServerSpec extends AnyFlatSpec with Matchers: val req = (Request(method = "initialize", id = RequestId("i")): JSONRPCMessage).asJson val result = resultJson(handler.handleJsonRpc(req, Seq.empty)).as[InitializeResult].getOrElse(fail("decode InitializeResult")) result.capabilities.extensions.map(_.keySet) shouldBe Some(Set(TasksExtension.Id)) + + it should "run an input_required task tool: request input, answer via tasks/update, then complete" in: + val handler = handlerWith() + val created = resultJson(handler.handleJsonRpc(callToolReq("askThenEcho", withTasks = true), Seq.empty)) + .as[CreateTaskResult] + .getOrElse(fail("decode CreateTaskResult")) + created.status shouldBe TaskStatus.Working + + // the tool has requested input and parked, surfacing the request via tasks/get + val waiting = pollUntil(handler, created.taskId)(_.status == TaskStatus.InputRequired) + waiting.outcome match + case TaskOutcome.InputRequired(requests) => requests.keySet shouldBe Set("q1") + case other => fail(s"expected InputRequired, got $other") + + // answer it + val updateReq = (Request( + method = "tasks/update", + params = Some(UpdateTaskParams(created.taskId, Map("q1" -> Json.fromString("42"))).asJson), + id = RequestId("u") + ): JSONRPCMessage).asJson + val ack = resultJson(handler.handleJsonRpc(updateReq, Seq.empty)).as[TaskAck].getOrElse(fail("decode TaskAck")) + ack.taskId shouldBe Some(created.taskId) + + // the tool resumes and completes with the answer folded in + pollTask(handler, created.taskId).outcome match + case TaskOutcome.Completed(result) => + result.as[CallToolResult].toOption.map(_.content.head) shouldBe Some(ToolContent.Text("text", "got:\"42\"")) + case other => fail(s"expected Completed, got $other") + + it should "reject the input_required tool when the client does not declare task support" in: + val handler = handlerWith() + val err = errorObj(handler.handleJsonRpc(callToolReq("askThenEcho", withTasks = false), Seq.empty)) + err.code shouldBe JSONRPCErrorCodes.MissingRequiredClientCapability.code