Enrich server implementation (#171)

* feat: rich tool result set and server context

* feat: remove old mcp endpoint creation

* feat: DNS rebinding prevention, tests

* feat: server streaming abstractions and first implementation for ZIO

* refactor: general refactoring and renames

* refactor: clear close semantics for zio server streaming

* refactor: rename

* feat: very basic scala doc for public API, inline logic handlers across server methods

* feat: introduce stdio transport to server, use similar abstraction as for the client

* refactor: simpler comment

* refactor: comments

* format

* feat: harness stdio server tests

* renaming

* renaming

* stdio implementation for zio server

* feat: documentation updates

* remove comments

* simplify server context

* feat: simplify server context

* feat: resource not found instead of invalid params

* fix: scoped event stream

* fix: update readme

Author: kubinio123

README.md

@@ -5,7 +5,8 @@
 
 An SDK for building [MCP](https://modelcontextprotocol.io/specification) (Model Context Protocol) servers and
 clients in Scala 3 using boilerplate-less, type-safe APIs based on [Tapir](https://tapir.softwaremill.com/)
-and [sttp](https://github.com/softwaremill/sttp), supporting the variety of the Scala ecosystem.
+and [sttp](https://github.com/softwaremill/sttp), supporting the variety of the Scala ecosystem. Both servers
+and clients can communicate over **Streamable HTTP** or **stdio**.
 
 ### Quickstart
 
@@ -24,9 +25,9 @@ case class AdderInput(a: Int, b: Int) derives Codec, Schema
 
 @main def server(): Unit =
   val adder = tool("adder").description("Adds two numbers").input[AdderInput]
-    .handle(i => Right(s"Result: ${i.a + i.b}"))
+    .handle(i => ToolResult.text(s"Result: ${i.a + i.b}"))
 
-  NettySyncServer().port(8080).addEndpoint(mcpEndpoint(List(adder), List("mcp"))).startAndWait()
+  NettySyncServer().port(8080).addEndpoint(McpServer(tools = List(adder)).endpoint(List("mcp"))).startAndWait()
 ```
 
 Connect and invoke the tool as an MCP client:
@@ -36,14 +37,14 @@ Connect and invoke the tool as an MCP client:
 //> using dep com.softwaremill.sttp.client4::core:4.0.24
 
 import chimp.client.*
-import chimp.client.transport.HttpTransport
+import chimp.client.transport.ClientHttpTransport
 import chimp.protocol.*
 import sttp.client4.*
 import io.circe.Json
 
 @main def client(): Unit =
   val backend = DefaultSyncBackend()
-  val transport = HttpTransport(backend, uri"http://localhost:8080/mcp")
+  val transport = ClientHttpTransport(backend, uri"http://localhost:8080/mcp")
   val client = McpClient(transport, Implementation("my-client", "0.1.0"))
 
   val result = client.callTool("adder", Json.obj("a" -> Json.fromInt(2), "b" -> Json.fromInt(3)))

build.sbt

@@ -10,6 +10,7 @@ val tapirV = "1.13.26"
 val sttpClientV = "4.0.26"
 val zioV = "2.1.26"
 val zioProcessV = "0.8.0"
+val zioHttpV = "3.8.0"
 val testcontainersScalaV = "0.41.8"
 
 lazy val verifyExamplesCompileUsingScalaCli = taskKey[Unit]("Verify that each example compiles using Scala CLI")
@@ -35,7 +36,7 @@ val scalaTest = "org.scalatest" %% "scalatest" % scalaTestV % Test
 lazy val root = (project in file("."))
   .settings(commonSettings: _*)
   .settings(publishArtifact := false, name := "chimp")
-  .aggregate(core, server, client, clientZio, examples, serverConformance, clientConformance)
+  .aggregate(core, server, serverZio, client, clientZio, examples, serverConformance, clientConformance)
 
 val conformance = inputKey[Unit]("Run the MCP conformance harness via npx, extra args are passed through")
 
@@ -62,10 +63,27 @@ lazy val server: Project = (project in file("server"))
       "com.softwaremill.sttp.tapir" %% "tapir-core" % tapirV,
       "com.softwaremill.sttp.tapir" %% "tapir-json-circe" % tapirV,
       "com.softwaremill.sttp.tapir" %% "tapir-apispec-docs" % tapirV,
-      "com.softwaremill.sttp.apispec" %% "jsonschema-circe" % "0.11.10"
+      "com.softwaremill.sttp.apispec" %% "jsonschema-circe" % "0.11.10",
+      "com.softwaremill.sttp.tapir" %% "tapir-netty-server-sync" % tapirV % Test,
+      "com.softwaremill.sttp.client4" %% "core" % sttpClientV % Test
     )
   )
-  .dependsOn(core)
+  .dependsOn(core, client % "test->compile")
+
+lazy val serverZio: Project = (project in file("server-streaming/server-zio"))
+  .settings(commonSettings: _*)
+  .settings(
+    name := "chimp-server-zio",
+    libraryDependencies ++= Seq(
+      scalaTest,
+      "dev.zio" %% "zio" % zioV,
+      "dev.zio" %% "zio-streams" % zioV,
+      "com.softwaremill.sttp.tapir" %% "tapir-zio" % tapirV,
+      "com.softwaremill.sttp.tapir" %% "tapir-zio-http-server" % tapirV,
+      "dev.zio" %% "zio-http" % zioHttpV
+    )
+  )
+  .dependsOn(server % "compile->compile;test->test", clientZio % "test->compile")
 
 lazy val client: Project = (project in file("client"))
   .settings(commonSettings: _*)
@@ -244,7 +262,8 @@ lazy val docs: Project = (project in file("generated-docs"))
     ),
     mdocOut := file("generated-docs/out"),
     mdocExtraArguments := Seq("--clean-target", "--exclude", ".venv", "--exclude", "_build"),
+    libraryDependencies += "com.softwaremill.sttp.tapir" %% "tapir-netty-server-sync" % tapirV,
     publishArtifact := false,
     name := "docs"
   )
-  .dependsOn(core, server, client, clientZio)
+  .dependsOn(core, server, serverZio, client, clientZio)

client-conformance/src/main/scala/chimp/conformance/client/Main.scala

@@ -1,7 +1,7 @@
 package chimp.conformance.client
 
 import chimp.client.McpClient
-import chimp.client.transport.HttpTransport
+import chimp.client.transport.ClientHttpTransport
 import chimp.protocol.*
 import io.circe.Json
 import sttp.client4.DefaultSyncBackend
@@ -28,7 +28,7 @@ object Main:
       .getOrElse(ProtocolVersion.Latest)
 
     val backend = DefaultSyncBackend()
-    val transport = HttpTransport[Identity](backend, serverUrl, protocolVersion)
+    val transport = ClientHttpTransport[Identity](backend, serverUrl, protocolVersion)
 
     val rc: Int =
       try

…port/zio/ZioStreamingHttpTransport.scala …ansport/zio/ZioClientHttpTransport.scalaclient-streaming/client-zio/src/main/scala/chimp/client/transport/zio/ZioStreamingHttpTransport.scala renamed to client-streaming/client-zio/src/main/scala/chimp/client/transport/zio/ZioClientHttpTransport.scala client-streaming/client-zio/src/main/scala/chimp/client/transport/zio/ZioStreamingHttpTransport.scala renamed to client-streaming/client-zio/src/main/scala/chimp/client/transport/zio/ZioClientHttpTransport.scala

@@ -1,7 +1,7 @@
 package chimp.client.transport.zio
 
-import chimp.client.transport.HttpTransport.HttpOutcome
-import chimp.client.transport.{HttpTransport, StreamingHttpTransport, Transport}
+import chimp.client.transport.ClientHttpTransport.HttpOutcome
+import chimp.client.transport.{ClientHttpTransport, ClientStreamingHttpTransport, ClientTransport}
 import chimp.client.{McpProtocolException, McpSessionNotFoundException}
 import chimp.protocol.{JSONRPCErrorCodes, JSONRPCErrorObject, JSONRPCMessage, ProtocolVersion, RequestId}
 import org.slf4j.LoggerFactory
@@ -15,7 +15,7 @@ import zio.{Duration, Exit, Promise, Ref, Schedule, Scope, Task, ZIO, ZLayer}
 
 import scala.concurrent.duration.FiniteDuration
 
-final class ZioStreamingHttpTransport private (
+final class ZioClientHttpTransport private (
     backend: StreamBackend[Task, ZioStreams],
     uri: Uri,
     protocolVersion: ProtocolVersion,
@@ -28,9 +28,9 @@ final class ZioStreamingHttpTransport private (
     incomingRef: Ref[JSONRPCMessage => Task[Unit]],
     lastEventId: Ref[Option[String]],
     closingRef: Ref[Boolean]
-) extends StreamingHttpTransport[Task, ZioStreams](backend, uri, ZioStreams):
+) extends ClientStreamingHttpTransport[Task, ZioStreams](backend, uri, ZioStreams):
 
-  private val log = LoggerFactory.getLogger(classOf[ZioStreamingHttpTransport])
+  private val log = LoggerFactory.getLogger(classOf[ZioClientHttpTransport])
 
   override given monad: MonadError[Task] = backend.monad
 
@@ -53,7 +53,7 @@ final class ZioStreamingHttpTransport private (
       .getAndSet(None)
       .flatMap:
         case Some(id) =>
-          HttpTransport
+          ClientHttpTransport
             .baseDeleteRequest(uri, protocolVersion, id)
             .response(asStreamUnsafe(ZioStreams))
             .send(backend)
@@ -66,7 +66,7 @@ final class ZioStreamingHttpTransport private (
     post(request).flatMap: resp =>
       captureSession(resp) *>
         sessionRef.get.flatMap: session =>
-          HttpTransport.resolveResponse(resp, session) match
+          ClientHttpTransport.resolveResponse(resp, session) match
             case Left(err: McpSessionNotFoundException) =>
               sessionRef.set(None) *> ZIO.fail(err)
             case Left(err) =>
@@ -87,7 +87,7 @@ final class ZioStreamingHttpTransport private (
     post(msg).flatMap: response =>
       captureSession(response) *>
         sessionRef.get.flatMap: session =>
-          HttpTransport.resolveResponse(response, session) match
+          ClientHttpTransport.resolveResponse(response, session) match
             case Left(err: McpSessionNotFoundException) =>
               sessionRef.set(None) *> ZIO.fail(err)
             case Left(err) =>
@@ -101,8 +101,8 @@ final class ZioStreamingHttpTransport private (
 
   private def post(msg: JSONRPCMessage): Task[Response[Either[String, Stream[Throwable, Byte]]]] =
     sessionRef.get.flatMap: session =>
-      HttpTransport
-        .basePostRequest(uri, protocolVersion, session, Transport.encode(msg))
+      ClientHttpTransport
+        .basePostRequest(uri, protocolVersion, session, ClientTransport.encode(msg))
         .response(asStreamUnsafe(ZioStreams))
         .send(backend)
 
@@ -123,7 +123,7 @@ final class ZioStreamingHttpTransport private (
       case Right(stream) => stream.runDrain.ignore
 
   private def decode(body: String): Task[JSONRPCMessage] =
-    Transport.decode(body) match
+    ClientTransport.decode(body) match
       case Right(msg) => ZIO.succeed(msg)
       case Left(err)  => ZIO.fail(McpProtocolException(s"Failed to decode response body: ${err.getMessage}, payload $body"))
 
@@ -164,7 +164,7 @@ final class ZioStreamingHttpTransport private (
   private def dispatch(event: ServerSentEvent): Task[Unit] =
     event.data match
       case Some(data) if data.nonEmpty =>
-        Transport.decode(data) match
+        ClientTransport.decode(data) match
           case Right(msg) => routeMessage(msg)
           case Left(_)    => ZIO.unit
       case _ => ZIO.unit
@@ -257,7 +257,7 @@ final class ZioStreamingHttpTransport private (
       error = JSONRPCErrorObject(code = JSONRPCErrorCodes.InvocationError.code, message = "SSE stream ended before response")
     )
 
-object ZioStreamingHttpTransport:
+object ZioClientHttpTransport:
 
   val defaultReconnectSchedule: Schedule[Any, Any, Any] =
     Schedule.exponential(Duration.fromMillis(100)).jittered || Schedule.spaced(Duration.fromSeconds(30))
@@ -266,9 +266,9 @@ object ZioStreamingHttpTransport:
       backend: StreamBackend[Task, ZioStreams],
       uri: Uri,
       protocolVersion: ProtocolVersion = ProtocolVersion.Latest,
-      timeout: FiniteDuration = Transport.defaultTimeout,
+      timeout: FiniteDuration = ClientTransport.defaultTimeout,
       reconnectSchedule: Schedule[Any, Any, Any] = defaultReconnectSchedule
-  ): Task[ZioStreamingHttpTransport] =
+  ): Task[ZioClientHttpTransport] =
     for
       scope <- Scope.make
       sessionRef <- Ref.make(Option.empty[String])
@@ -277,7 +277,7 @@ object ZioStreamingHttpTransport:
       incomingRef <- Ref.make[JSONRPCMessage => Task[Unit]](_ => ZIO.unit)
       lastEventId <- Ref.make(Option.empty[String])
       closingRef <- Ref.make(false)
-      transport = new ZioStreamingHttpTransport(
+      transport = new ZioClientHttpTransport(
         backend,
         uri,
         protocolVersion,
@@ -298,16 +298,16 @@ object ZioStreamingHttpTransport:
       backend: StreamBackend[Task, ZioStreams],
       uri: Uri,
       protocolVersion: ProtocolVersion = ProtocolVersion.Latest,
-      timeout: FiniteDuration = Transport.defaultTimeout,
+      timeout: FiniteDuration = ClientTransport.defaultTimeout,
       reconnectSchedule: Schedule[Any, Any, Any] = defaultReconnectSchedule
-  ): ZIO[Scope, Throwable, ZioStreamingHttpTransport] =
+  ): ZIO[Scope, Throwable, ZioClientHttpTransport] =
     ZIO.acquireRelease(apply(backend, uri, protocolVersion, timeout, reconnectSchedule))(_.close().ignore)
 
   def layer(
       backend: StreamBackend[Task, ZioStreams],
       uri: Uri,
       protocolVersion: ProtocolVersion = ProtocolVersion.Latest,
-      timeout: FiniteDuration = Transport.defaultTimeout,
+      timeout: FiniteDuration = ClientTransport.defaultTimeout,
       reconnectSchedule: Schedule[Any, Any, Any] = defaultReconnectSchedule
-  ): ZLayer[Any, Throwable, ZioStreamingHttpTransport] =
+  ): ZLayer[Any, Throwable, ZioClientHttpTransport] =
     ZLayer.scoped(scoped(backend, uri, protocolVersion, timeout, reconnectSchedule))

…ort/zio/ZioStreamingStdioTransport.scala …nsport/zio/ZioClientStdioTransport.scalaclient-streaming/client-zio/src/main/scala/chimp/client/transport/zio/ZioStreamingStdioTransport.scala renamed to client-streaming/client-zio/src/main/scala/chimp/client/transport/zio/ZioClientStdioTransport.scala client-streaming/client-zio/src/main/scala/chimp/client/transport/zio/ZioStreamingStdioTransport.scala renamed to client-streaming/client-zio/src/main/scala/chimp/client/transport/zio/ZioClientStdioTransport.scala

@@ -1,6 +1,6 @@
 package chimp.client.transport.zio
 
-import chimp.client.transport.{StreamingStdioTransport, Transport}
+import chimp.client.transport.{ClientStreamingStdioTransport, ClientTransport}
 import chimp.protocol.JSONRPCMessage
 import org.slf4j.LoggerFactory
 import sttp.client4.impl.zio.RIOMonadAsyncError
@@ -13,7 +13,7 @@ import java.io.File
 import java.nio.charset.StandardCharsets
 import scala.concurrent.duration.FiniteDuration
 
-final class ZioStreamingStdioTransport private (
+final class ZioClientStdioTransport private (
     command: List[String],
     env: Map[String, String],
     workDir: Option[File],
@@ -23,9 +23,9 @@ final class ZioStreamingStdioTransport private (
     writeQueue: Queue[JSONRPCMessage],
     pending: ZioPendingRequests,
     incomingRef: Ref[JSONRPCMessage => Task[Unit]]
-) extends StreamingStdioTransport[Task](command, env, workDir):
+) extends ClientStreamingStdioTransport[Task](command, env, workDir):
 
-  private val log = LoggerFactory.getLogger(classOf[ZioStreamingStdioTransport])
+  private val log = LoggerFactory.getLogger(classOf[ZioClientStdioTransport])
 
   override given monad: MonadError[Task] = new RIOMonadAsyncError[Any]
 
@@ -54,7 +54,7 @@ final class ZioStreamingStdioTransport private (
     val drain = process.stdout.linesStream
       .filter(_.nonEmpty)
       .mapZIO: line =>
-        Transport.decode(line) match
+        ClientTransport.decode(line) match
           case Right(msg) => dispatch(msg)
           case Left(err)  => ZIO.succeed(log.warn(s"Failed to parse JSON-RPC line: ${err.getMessage}, raw: $line"))
       .runDrain
@@ -68,29 +68,29 @@ final class ZioStreamingStdioTransport private (
       .forkIn(scope)
       .unit
 
-object ZioStreamingStdioTransport:
+object ZioClientStdioTransport:
 
   def apply(
       command: List[String],
       env: Map[String, String] = Map.empty,
       workDir: Option[File] = None,
-      timeout: FiniteDuration = Transport.defaultTimeout
-  ): Task[ZioStreamingStdioTransport] =
+      timeout: FiniteDuration = ClientTransport.defaultTimeout
+  ): Task[ZioClientStdioTransport] =
     for
       scope <- Scope.make
       writeQueue <- Queue.bounded[JSONRPCMessage](256)
       pending <- ZioPendingRequests.make
       incomingRef <- Ref.make[JSONRPCMessage => Task[Unit]](_ => ZIO.unit)
       stdinBytes = ZStream
         .fromQueue(writeQueue)
-        .map(msg => Chunk.fromArray((Transport.encode(msg) + "\n").getBytes(StandardCharsets.UTF_8)))
+        .map(msg => Chunk.fromArray((ClientTransport.encode(msg) + "\n").getBytes(StandardCharsets.UTF_8)))
         .flattenChunks
       baseCmd = Command(command.head, command.tail*)
       withEnv = if env.isEmpty then baseCmd else baseCmd.env(env)
       withDir = workDir.fold(withEnv)(withEnv.workingDirectory)
       cmd = withDir.stdin(ProcessInput.fromStream(stdinBytes, flushChunksEagerly = true))
       process <- cmd.run.provideEnvironment(zio.ZEnvironment(scope))
-      transport = new ZioStreamingStdioTransport(command, env, workDir, timeout, scope, process, writeQueue, pending, incomingRef)
+      transport = new ZioClientStdioTransport(command, env, workDir, timeout, scope, process, writeQueue, pending, incomingRef)
       _ <- transport.startReader
       _ <- transport.startStderr
     yield transport
@@ -99,14 +99,14 @@ object ZioStreamingStdioTransport:
       command: List[String],
       env: Map[String, String] = Map.empty,
       workDir: Option[File] = None,
-      timeout: FiniteDuration = Transport.defaultTimeout
-  ): ZIO[Scope, Throwable, ZioStreamingStdioTransport] =
+      timeout: FiniteDuration = ClientTransport.defaultTimeout
+  ): ZIO[Scope, Throwable, ZioClientStdioTransport] =
     ZIO.acquireRelease(apply(command, env, workDir, timeout))(_.close().ignore)
 
   def layer(
       command: List[String],
       env: Map[String, String] = Map.empty,
       workDir: Option[File] = None,
-      timeout: FiniteDuration = Transport.defaultTimeout
-  ): ZLayer[Any, Throwable, ZioStreamingStdioTransport] =
+      timeout: FiniteDuration = ClientTransport.defaultTimeout
+  ): ZLayer[Any, Throwable, ZioClientStdioTransport] =
     ZLayer.scoped(scoped(command, env, workDir, timeout))

…io/ZioStreamingHttpIntegrationSpec.scala …io/ZioMcpClientHttpIntegrationSpec.scalaclient-streaming/client-zio/src/test/scala/chimp/client/transport/zio/ZioStreamingHttpIntegrationSpec.scala renamed to client-streaming/client-zio/src/test/scala/chimp/client/transport/zio/ZioMcpClientHttpIntegrationSpec.scala client-streaming/client-zio/src/test/scala/chimp/client/transport/zio/ZioStreamingHttpIntegrationSpec.scala renamed to client-streaming/client-zio/src/test/scala/chimp/client/transport/zio/ZioMcpClientHttpIntegrationSpec.scala

@@ -1,7 +1,7 @@
 package chimp.client.transport.zio
 
-import chimp.client.integration.StreamingHttpIntegrationSpec
-import chimp.client.transport.BidirectionalTransport
+import chimp.client.integration.McpClientStreamingHttpIntegrationSpec
+import chimp.client.transport.ClientBidirectionalTransport
 import chimp.protocol.ProtocolVersion
 import sttp.capabilities.zio.ZioStreams
 import sttp.client4.StreamBackend
@@ -11,13 +11,13 @@ import zio.{Task, ZIO}
 
 import scala.concurrent.duration.FiniteDuration
 
-class ZioStreamingHttpIntegrationSpec extends StreamingHttpIntegrationSpec[Task, StreamBackend[Task, ZioStreams]] with ZioToFuture:
+class ZioMcpClientHttpIntegrationSpec extends McpClientStreamingHttpIntegrationSpec[Task, StreamBackend[Task, ZioStreams]] with ZioToFuture:
 
   override def usingBackend[A](use: StreamBackend[Task, ZioStreams] => Task[A]): Task[A] =
     HttpClientZioBackend().flatMap: b =>
       use(b).ensuring(b.close().orDie)
 
   override def usingBidirectionalTransport[A](b: StreamBackend[Task, ZioStreams], uri: Uri, timeout: FiniteDuration)(
-      use: BidirectionalTransport[Task] => Task[A]
+      use: ClientBidirectionalTransport[Task] => Task[A]
   ): Task[A] =
-    ZIO.scoped(ZioStreamingHttpTransport.scoped(b, uri, ProtocolVersion.Latest, timeout).flatMap(use))
+    ZIO.scoped(ZioClientHttpTransport.scoped(b, uri, ProtocolVersion.Latest, timeout).flatMap(use))

client-streaming/client-zio/src/test/scala/chimp/client/transport/zio/ZioMcpClientStdioIntegrationSpec.scala

@@ -0,0 +1,14 @@
+package chimp.client.transport.zio
+
+import chimp.client.integration.McpClientStdioIntegrationSpec
+import chimp.client.transport.ClientBidirectionalTransport
+import zio.{Task, ZIO}
+
+import scala.concurrent.duration.FiniteDuration
+
+class ZioMcpClientStdioIntegrationSpec extends McpClientStdioIntegrationSpec[Task] with ZioToFuture:
+
+  override def usingTransport[A](command: List[String], timeout: FiniteDuration)(
+      use: ClientBidirectionalTransport[Task] => Task[A]
+  ): Task[A] =
+    ZIO.scoped(ZioClientStdioTransport.scoped(command, timeout = timeout).flatMap(use))