Add serverSecurityLogic

docs/server/tools.md

@@ -8,8 +8,9 @@
   - `handle` — synchronous logic from input to `ToolResult`.
   - `handleWithHeaders` — synchronous logic that also receives the request headers.
   - `serverLogic` — effectful logic, with the request headers.
+  - `handleSecured` (synchronous) or `securedServerLogic` (effectful) — logic that also receives the principal made by the server's security logic; usable only on a secured server — see [transport security](transport.md).
 
-  A tool that pushes to the client while running (progress, logging) instead uses `streamingServerLogic` — see [server capabilities](capabilities.md).
+  A tool that pushes to the client while running (progress, logging) instead uses `streamingServerLogic` — see [server capabilities](capabilities.md). A secured, streaming tool uses `securedStreamingServerLogic`, which gives both the principal and the streaming context — see [transport security](transport.md).
 - Assemble tools into an `McpServer` and call `.endpoint(path)` to create a Tapir endpoint.
 
 ```scala mdoc:compile-only

docs/server/transport.md

@@ -106,6 +106,82 @@ object SecuredMcpServer:
     NettySyncServer().port(8080).addEndpoint(securedEndpoint).startAndWait()
 ```
 
-The result of the security logic does not reach the tool logic. If a tool needs data from the caller, read the request headers with `handleWithHeaders` (or `serverLogic` with headers).
+The result of the security logic of `prependSecurity` does not reach the tool logic. Use `prependSecurity` if the tools do not need data from the caller. If a tool needs such data, read the request headers with `handleWithHeaders` (or `serverLogic` with headers), or give the tools a principal, as below.
 
 For all the security inputs - API keys, basic and bearer authorization, OAuth2 flows - see the [Tapir endpoint security documentation](https://tapir.softwaremill.com/en/latest/endpoint/security.html).
+
+### Giving the security result to the tools
+
+To validate the caller one time and give the result to the tool logic, use `serverSecurityLogic` (or `serverSecurityLogicPure`, if the logic needs no effect). It takes the same security input and error output as `prependSecurity`, and makes a principal - a value of your own type, such as the identity of the caller.
+
+The server gives the principal to the logic of each tool which you add to it. Define such a tool with `handleSecured`, or with `securedServerLogic` if the logic needs an effect. Tools which do not need the principal keep their usual logic. The other builders of `McpServer` stay available, so you can configure the server before or after you add the security logic:
+
+```scala mdoc:compile-only
+import chimp.server.*
+import sttp.model.StatusCode
+import sttp.shared.Identity
+import sttp.tapir.*
+import sttp.tapir.server.netty.sync.NettySyncServer
+
+case class User(email: String)
+
+object McpServerWithPrincipal:
+  def main(args: Array[String]): Unit =
+    val echo = tool("echo").input[String].handle(message => ToolResult.text(message))
+    val whoAmI = tool("whoAmI").input[String].handleSecured[User]((_, user) => ToolResult.text(user.email))
+
+    val securedEndpoint = McpServer[Identity]()
+      .serverSecurityLogicPure(
+        auth.bearer[String](),
+        statusCode(StatusCode.Unauthorized).and(stringBody)
+      )(token => if token == "s3cret" then Right(User("employee@example.com")) else Left("Invalid token"))
+      .name("my-mcp-server")
+      .addTools(echo, whoAmI)
+      .endpoint(List("mcp"))
+
+    NettySyncServer().port(8080).addEndpoint(securedEndpoint).startAndWait()
+```
+
+The security logic runs one time for each request, before the server handles the MCP message. If it gives a rejection, the server sends the error output and no tool logic runs. This is necessary if the client must get an HTTP status code, because a tool which rejects a call can only give a JSON-RPC error with status code 200.
+
+### Combining security with streaming
+
+`.streaming` on a secured server gives a `SecuredStreamingMcpServer`, which accepts streaming tools alongside the plain secured ones. Define such a tool with `securedStreamingServerLogic`, which gives both the principal and the `StreamingServerContext`:
+
+```scala mdoc:compile-only
+import chimp.protocol.LoggingLevel
+import chimp.server.*
+import chimp.server.ox.OxServerHttpTransport
+import chimp.server.transport.SecuredServerStreamingHttpTransport
+import io.circe.Json
+import sttp.model.StatusCode
+import sttp.shared.Identity
+import sttp.tapir.*
+import sttp.tapir.server.netty.sync.NettySyncServer
+
+case class User(email: String)
+
+object McpServerWithPrincipalAndStreaming:
+  def main(args: Array[String]): Unit =
+    val whoAmI = tool("whoAmI")
+      .input[String]
+      .securedStreamingServerLogic[Identity, User]: (_, user, ctx, _) =>
+        ctx.log(LoggingLevel.Info, Json.fromString(s"called by ${user.email}"))
+        ToolResult.text(user.email)
+
+    val securedStreamingServer = McpServer[Identity]()
+      .serverSecurityLogicPure(
+        auth.bearer[String](),
+        statusCode(StatusCode.Unauthorized).and(stringBody)
+      )(token => if token == "s3cret" then Right(User("employee@example.com")) else Left("Invalid token"))
+      .streaming
+      .addStreamingTool(whoAmI)
+
+    // `OxServerHttpTransport` is one effect backend's streaming machinery; substitute your own (Pekko, ZIO, ...).
+    val backend = OxServerHttpTransport(List("mcp"))
+    val securedEndpoint = SecuredServerStreamingHttpTransport(List("mcp"), backend).serve(securedStreamingServer)
+
+    NettySyncServer().port(8080).addEndpoint(securedEndpoint).startAndWait()
+```
+
+`SecuredServerStreamingHttpTransport` takes an existing `ServerStreamingHttpTransport` as its source of streaming machinery, rather than extending it, so the same effect-specific backend instance - `OxServerHttpTransport`, `PekkoServerHttpTransport`, `ZioServerHttpTransport` - serves both a plain `StreamingMcpServer` and a `SecuredStreamingMcpServer`.

server-streaming/server-ox/src/test/scala/chimp/server/ox/OxMcpServerHttpSpec.scala

@@ -1,20 +1,39 @@
 package chimp.server.ox
 
-import chimp.client.transport.ClientTransport
+import chimp.client.transport.{ClientHttpTransport, ClientTransport}
 import chimp.client.transport.ox.OxClientHttpTransport
 import chimp.client.{BidirectionalMcpClient, McpClient}
 import chimp.protocol.{Implementation, ProtocolVersion}
-import chimp.server.{McpServer, McpServerStreamingTests, McpServerTests, StreamingMcpServer, SyncToFuture}
+import chimp.server.transport.SecuredServerStreamingHttpTransport
+import chimp.server.{
+  McpServer,
+  McpServerStreamingTests,
+  McpServerTests,
+  SecuredMcpServer,
+  SecuredMcpServerStreamingTests,
+  SecuredMcpServerTests,
+  SecuredStreamingMcpServer,
+  StreamingMcpServer,
+  SyncToFuture
+}
 import org.scalatest.Assertion
-import ox.supervised
-import sttp.client4.DefaultSyncBackend
+import ox.{supervised, Ox}
+import sttp.capabilities.WebSockets
+import sttp.client4.{DefaultSyncBackend, SyncBackend}
+import sttp.model.Header
 import sttp.model.Uri.UriContext
 import sttp.shared.Identity
-import sttp.tapir.server.netty.sync.NettySyncServer
+import sttp.tapir.server.netty.sync.{NettySyncServer, OxStreams}
+import sttp.tapir.server.ServerEndpoint
 
 import scala.concurrent.Future
 
-class OxMcpServerHttpSpec extends McpServerTests[Identity] with McpServerStreamingTests[Identity] with SyncToFuture:
+class OxMcpServerHttpSpec
+    extends McpServerTests[Identity]
+    with McpServerStreamingTests[Identity]
+    with SecuredMcpServerTests[Identity]
+    with SecuredMcpServerStreamingTests[Identity]
+    with SyncToFuture:
   private val clientInfo = Implementation("chimp-server-test", "0.0.1")
 
   override protected def withServer(server: McpServer[Identity])(test: McpClient[Identity] => Identity[Assertion]): Future[Assertion] =
@@ -23,21 +42,51 @@ class OxMcpServerHttpSpec extends McpServerTests[Identity] with McpServerStreami
   override protected def withStreamingServer(
       server: StreamingMcpServer[Identity]
   )(test: BidirectionalMcpClient[Identity] => Identity[Assertion]): Future[Assertion] =
+    withHttpServer(OxServerHttpTransport(List("mcp")).serve(server)): (port, backend, ox) =>
+      given Ox = ox
+      val transport =
+        OxClientHttpTransport(backend, uri"http://localhost:$port/mcp", ProtocolVersion.Latest, ClientTransport.defaultTimeout)
+      try test(McpClient.bidirectional(transport, clientInfo))
+      finally transport.close()
+
+  override protected def withSecuredServer(
+      server: SecuredMcpServer[Identity, String, String, User],
+      token: String
+  )(test: McpClient[Identity] => Identity[Assertion]): Future[Assertion] =
+    withHttpServer(server.endpoint(List("mcp"))): (port, backend, _) =>
+      val transport = ClientHttpTransport[Identity](
+        backend,
+        uri"http://localhost:$port/mcp",
+        headers = List(Header.authorization("Bearer", token))
+      )
+      try test(McpClient(transport, clientInfo))
+      finally transport.close()
+
+  override protected def withSecuredStreamingServer(
+      server: SecuredStreamingMcpServer[Identity, String, String, User],
+      token: String
+  )(test: BidirectionalMcpClient[Identity] => Identity[Assertion]): Future[Assertion] =
+    val endpoint = SecuredServerStreamingHttpTransport(List("mcp"), OxServerHttpTransport(List("mcp"))).serve(server)
+    withHttpServer(endpoint): (port, backend, ox) =>
+      given Ox = ox
+      val transport = OxClientHttpTransport(
+        backend,
+        uri"http://localhost:$port/mcp",
+        ProtocolVersion.Latest,
+        ClientTransport.defaultTimeout,
+        headers = List(Header.authorization("Bearer", token))
+      )
+      try test(McpClient.bidirectional(transport, clientInfo))
+      finally transport.close()
+
+  private def withHttpServer(
+      endpoint: ServerEndpoint[OxStreams & WebSockets, Identity]
+  )(test: (Int, SyncBackend, Ox) => Assertion): Future[Assertion] =
     toFuture:
       supervised:
-        val endpoint = OxServerHttpTransport(List("mcp")).serve(server)
         val binding = NettySyncServer().port(0).addEndpoint(endpoint).start()
         try
           val backend = DefaultSyncBackend()
-          try
-            val transport =
-              OxClientHttpTransport(
-                backend,
-                uri"http://localhost:${binding.port}/mcp",
-                ProtocolVersion.Latest,
-                ClientTransport.defaultTimeout
-              )
-            try test(McpClient.bidirectional(transport, clientInfo))
-            finally transport.close()
+          try test(binding.port, backend, summon[Ox])
           finally backend.close()
         finally binding.stop()

server-streaming/server-pekko/src/test/scala/chimp/server/pekko/PekkoMcpServerHttpSpec.scala

@@ -3,17 +3,36 @@ package chimp.server.pekko
 import chimp.client.transport.pekko.PekkoClientHttpTransport
 import chimp.client.{BidirectionalMcpClient, McpClient}
 import chimp.protocol.Implementation
-import chimp.server.{McpServer, McpServerStreamingTests, McpServerTests, StreamingMcpServer}
+import chimp.server.transport.SecuredServerStreamingHttpTransport
+import chimp.server.{
+  McpServer,
+  McpServerStreamingTests,
+  McpServerTests,
+  SecuredMcpServer,
+  SecuredMcpServerStreamingTests,
+  SecuredMcpServerTests,
+  SecuredStreamingMcpServer,
+  StreamingMcpServer
+}
 import org.apache.pekko.http.scaladsl.Http
 import org.scalatest.Assertion
+import sttp.capabilities.WebSockets
+import sttp.capabilities.pekko.PekkoStreams
 import sttp.client4.pekkohttp.PekkoHttpBackend
+import sttp.model.Header
 import sttp.model.Uri.UriContext
 import sttp.tapir.server.pekkohttp.PekkoHttpServerInterpreter
+import sttp.tapir.server.ServerEndpoint
 
 import scala.concurrent.duration.DurationInt
 import scala.concurrent.{ExecutionContext, Future}
 
-class PekkoMcpServerHttpSpec extends McpServerTests[Future] with McpServerStreamingTests[Future] with PekkoToFuture:
+class PekkoMcpServerHttpSpec
+    extends McpServerTests[Future]
+    with McpServerStreamingTests[Future]
+    with SecuredMcpServerTests[Future]
+    with SecuredMcpServerStreamingTests[Future]
+    with PekkoToFuture:
   private val clientInfo = Implementation("chimp-server-test", "0.0.1")
 
   override protected def withServer(server: McpServer[Future])(test: McpClient[Future] => Future[Assertion]): Future[Assertion] =
@@ -22,17 +41,37 @@ class PekkoMcpServerHttpSpec extends McpServerTests[Future] with McpServerStream
   override protected def withStreamingServer(
       server: StreamingMcpServer[Future]
   )(test: BidirectionalMcpClient[Future] => Future[Assertion]): Future[Assertion] =
+    withHttpServer(PekkoServerHttpTransport(List("mcp")).serve(server), Nil): transport =>
+      McpClient.bidirectional(transport, clientInfo).flatMap(test)
+
+  override protected def withSecuredServer(
+      server: SecuredMcpServer[Future, String, String, User],
+      token: String
+  )(test: McpClient[Future] => Future[Assertion]): Future[Assertion] =
+    withHttpServer(server.endpoint(List("mcp")), List(Header.authorization("Bearer", token))): transport =>
+      McpClient(transport, clientInfo).flatMap(test)
+
+  override protected def withSecuredStreamingServer(
+      server: SecuredStreamingMcpServer[Future, String, String, User],
+      token: String
+  )(test: BidirectionalMcpClient[Future] => Future[Assertion]): Future[Assertion] =
+    val endpoint = SecuredServerStreamingHttpTransport(List("mcp"), PekkoServerHttpTransport(List("mcp"))).serve(server)
+    withHttpServer(endpoint, List(Header.authorization("Bearer", token))): transport =>
+      McpClient.bidirectional(transport, clientInfo).flatMap(test)
+
+  private def withHttpServer(
+      endpoint: ServerEndpoint[PekkoStreams & WebSockets, Future],
+      headers: List[Header]
+  )(test: PekkoClientHttpTransport => Future[Assertion]): Future[Assertion] =
     given ExecutionContext = actorSystem.dispatcher
-    val endpoint = PekkoServerHttpTransport(List("mcp")).serve(server)
     Http()
       .newServerAt("localhost", 0)
       .bind(PekkoHttpServerInterpreter().toRoute(endpoint))
       .flatMap: binding =>
         val backend = PekkoHttpBackend.usingActorSystem(actorSystem)
-        val transport = PekkoClientHttpTransport(backend, uri"http://localhost:${binding.localAddress.getPort}/mcp")
-        McpClient
-          .bidirectional(transport, clientInfo)
-          .flatMap(test)
+        val transport =
+          PekkoClientHttpTransport(backend, uri"http://localhost:${binding.localAddress.getPort}/mcp", headers = headers)
+        test(transport)
           .transformWith: result =>
             transport
               .close()

server-streaming/server-zio/src/test/scala/chimp/server/zio/ZioMcpServerHttpSpec.scala

@@ -4,17 +4,36 @@ import chimp.client.transport.ClientTransport
 import chimp.client.transport.zio.ZioClientHttpTransport
 import chimp.client.{BidirectionalMcpClient, McpClient}
 import chimp.protocol.{Implementation, ProtocolVersion}
-import chimp.server.{McpServer, McpServerStreamingTests, McpServerTests, StreamingMcpServer}
+import chimp.server.transport.SecuredServerStreamingHttpTransport
+import chimp.server.{
+  McpServer,
+  McpServerStreamingTests,
+  McpServerTests,
+  SecuredMcpServer,
+  SecuredMcpServerStreamingTests,
+  SecuredMcpServerTests,
+  SecuredStreamingMcpServer,
+  StreamingMcpServer
+}
 import org.scalatest.Assertion
+import sttp.capabilities.WebSockets
+import sttp.capabilities.zio.ZioStreams
 import sttp.client4.*
 import sttp.client4.httpclient.zio.HttpClientZioBackend
+import sttp.model.Header
 import sttp.tapir.server.ziohttp.ZioHttpInterpreter
+import sttp.tapir.server.ServerEndpoint
 import zio.http.Server
 import zio.{Scope, Task, ZIO}
 
 import scala.concurrent.Future
 
-class ZioMcpServerHttpSpec extends McpServerTests[Task] with McpServerStreamingTests[Task] with ZioToFuture:
+class ZioMcpServerHttpSpec
+    extends McpServerTests[Task]
+    with McpServerStreamingTests[Task]
+    with SecuredMcpServerTests[Task]
+    with SecuredMcpServerStreamingTests[Task]
+    with ZioToFuture:
   private val clientInfo = Implementation("chimp-server-test", "0.0.1")
 
   override protected def withServer(server: McpServer[Task])(test: McpClient[Task] => Task[Assertion]): Future[Assertion] =
@@ -23,15 +42,42 @@ class ZioMcpServerHttpSpec extends McpServerTests[Task] with McpServerStreamingT
   override protected def withStreamingServer(
       server: StreamingMcpServer[Task]
   )(test: BidirectionalMcpClient[Task] => Task[Assertion]): Future[Assertion] =
+    withHttpServer(ZioServerHttpTransport(List("mcp")).serve(server), Nil): transport =>
+      McpClient.bidirectional(transport, clientInfo).flatMap(test)
+
+  override protected def withSecuredServer(
+      server: SecuredMcpServer[Task, String, String, User],
+      token: String
+  )(test: McpClient[Task] => Task[Assertion]): Future[Assertion] =
+    withHttpServer(server.endpoint(List("mcp")), List(Header.authorization("Bearer", token))): transport =>
+      McpClient(transport, clientInfo).flatMap(test)
+
+  override protected def withSecuredStreamingServer(
+      server: SecuredStreamingMcpServer[Task, String, String, User],
+      token: String
+  )(test: BidirectionalMcpClient[Task] => Task[Assertion]): Future[Assertion] =
+    val endpoint = SecuredServerStreamingHttpTransport(List("mcp"), ZioServerHttpTransport(List("mcp"))).serve(server)
+    withHttpServer(endpoint, List(Header.authorization("Bearer", token))): transport =>
+      McpClient.bidirectional(transport, clientInfo).flatMap(test)
+
+  private def withHttpServer(
+      endpoint: ServerEndpoint[ZioStreams & WebSockets, Task],
+      headers: List[Header]
+  )(test: ZioClientHttpTransport => Task[Assertion]): Future[Assertion] =
     toFuture:
-      val routes = ZioHttpInterpreter().toHttp(ZioServerHttpTransport(List("mcp")).serve(server))
+      val routes = ZioHttpInterpreter().toHttp(endpoint)
       ZIO.scoped:
         (for
           port <- Server.install(routes)
           result <- HttpClientZioBackend().flatMap: backend =>
             ZioClientHttpTransport
-              .scoped(backend, uri"http://localhost:$port/mcp", ProtocolVersion.Latest, ClientTransport.defaultTimeout)
-              .flatMap(transport => McpClient.bidirectional(transport, clientInfo))
-              .flatMap(client => test(client))
+              .scoped(
+                backend,
+                uri"http://localhost:$port/mcp",
+                ProtocolVersion.Latest,
+                ClientTransport.defaultTimeout,
+                headers = headers
+              )
+              .flatMap(test)
               .ensuring(backend.close().ignore)
         yield result).provideSome[Scope](Server.defaultWithPort(0))