refactor(test): add centralized TypeScript server factory for integration tests (#477)

# refactor(test): add centralized TypeScript server factory for
integration tests

  - Create TypeScriptServer class for managing TypeScript MCP servers
  - Add dedicated typescript/ directory with server infrastructure
  - Refactor TsTestBase to use TypeScriptServer for stdio transport
- Extract TypeScript test tools (greet, multi-greet, prompts, resources)
  - Use npx tsx for direct execution (better performance than npm run)
  - Install TypeScript dependencies automatically in test setup

  The TypeScriptServer factory provides:
  - startSse(port): Start SSE transport server
  - startStdio(): Start STDIO transport server
  - stop(): Graceful server shutdown
  - installDependencies(): npm install automation

<!-- Provide a brief summary of your changes -->

## Motivation and Context
To resurrect typescript tests

## How Has This Been Tested?
Locally, CI

## Breaking Changes
<!-- Will users need to update their code or configurations? -->

## Types of changes
- [x] Bug fix (non-breaking change which fixes an issue)
- [x] Test fix

## Checklist
<!-- Go over all the following points, and put an `x` in all the boxes
that apply. -->
- [x] I have read the [MCP
Documentation](https://modelcontextprotocol.io)
- [x] My code follows the repository's style guidelines
- [x] New and existing tests pass locally
- [ ] I have added appropriate error handling
- [ ] I have added or updated documentation as needed

## Additional context
#475
This commit is contained in:
Konstantin Pavlov 2026-01-10 20:46:51 +02:00 • committed by GitHub
commit 658b370c87
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 2848 additions and 1119 deletions

View file

@ -41,7 +41,7 @@ jobs:
- os: windows-latest
job-name: "Windows Tests"
gradle-tasks: "jvmTest"
gradle-tasks: "jvmTest -x :integration-test:jvmTest"
max-workers: 3
- os: macos-latest
@ -64,6 +64,11 @@ jobs:
with:
node-version: 'lts/*'
- name: Set up Node Packages
if: matrix.os == 'ubuntu-latest'
working-directory: integration-test/src/jvmTest/typescript
run: npm ci
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v5
with:
@ -75,7 +80,7 @@ jobs:
caches
notifications
sdks
.konan/**
../.konan/**
- name: ${{ matrix.job-name }}
run: |-
@ -144,6 +149,11 @@ jobs:
with:
add-job-summary: 'always'
cache-read-only: true
gradle-home-cache-includes: |
caches
notifications
sdks
../.konan/**
- name: Download Maven Local Repository
uses: actions/download-artifact@v7

View file

@ -31,19 +31,24 @@ jobs:
- name: Setup Node.js
uses: actions/setup-node@v6
with:
node-version: '24.11.1'
node-version: 'lts/*'
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v5
with:
add-job-summary: 'always'
cache-read-only: true
cache-read-only: ${{ github.ref != 'refs/heads/main' }}
gradle-home-cache-includes: |
caches
notifications
sdks
../.konan/**
- name: Run Conformance Tests
run: ./gradlew --no-daemon :conformance-test:test
- name: Upload Conformance Results
if: ${{ !cancelled() }}
if: always()
uses: actions/upload-artifact@v6
with:
name: conformance-results

View file

@ -6,8 +6,8 @@ plugins {
dependencies {
testImplementation(project(":kotlin-sdk"))
testImplementation(kotlin("test"))
testImplementation(project(":test-utils"))
testImplementation(kotlin("test"))
testImplementation(libs.kotlin.logging)
testImplementation(libs.ktor.client.cio)
testImplementation(libs.ktor.server.cio)

View file

@ -1,6 +1,7 @@
package io.modelcontextprotocol.kotlin.sdk.conformance
import io.github.oshai.kotlinlogging.KotlinLogging
import io.modelcontextprotocol.kotlin.test.utils.findFreePort
import org.junit.jupiter.api.AfterAll
import org.junit.jupiter.api.BeforeAll
import org.junit.jupiter.api.DynamicTest
@ -10,7 +11,6 @@ import java.io.BufferedReader
import java.io.InputStreamReader
import java.lang.management.ManagementFactory
import java.net.HttpURLConnection
import java.net.ServerSocket
import java.net.URI
import java.util.concurrent.TimeUnit
import kotlin.properties.Delegates
@ -70,8 +70,6 @@ class ConformanceTest {
private const val GRACEFUL_SHUTDOWN_SECONDS = 5L
private const val FORCE_SHUTDOWN_SECONDS = 2L
private fun findFreePort(): Int = ServerSocket(0).use { it.localPort }
private fun getRuntimeClasspath(): String = ManagementFactory.getRuntimeMXBean().classPath
private fun getTestClasspath(): String = System.getProperty("test.classpath") ?: getRuntimeClasspath()

View file

@ -17,7 +17,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() {
protected abstract suspend fun <T> useClient(block: suspend (Client) -> T): T
@Test
@Timeout(30, unit = TimeUnit.SECONDS)
@Timeout(20, unit = TimeUnit.SECONDS)
fun connectsAndPings() = runBlocking(Dispatchers.IO) {
useClient { client ->
assertNotNull(client, "Client should be initialized")
@ -30,7 +30,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() {
}
@Test
@Timeout(30, unit = TimeUnit.SECONDS)
@Timeout(20, unit = TimeUnit.SECONDS)
fun listsTools() = runBlocking(Dispatchers.IO) {
useClient { client ->
val result = client.listTools()
@ -44,7 +44,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() {
}
@Test
@Timeout(30, unit = TimeUnit.SECONDS)
@Timeout(20, unit = TimeUnit.SECONDS)
fun callGreet() = runBlocking(Dispatchers.IO) {
useClient { client ->
val testName = "TestUser"
@ -59,7 +59,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() {
}
@Test
@Timeout(30, unit = TimeUnit.SECONDS)
@Timeout(20, unit = TimeUnit.SECONDS)
fun multipleClients() = runBlocking(Dispatchers.IO) {
useClient { client1 ->
useClient { client2 ->

View file

@ -11,59 +11,71 @@ import io.modelcontextprotocol.kotlin.sdk.server.Server
import io.modelcontextprotocol.kotlin.sdk.server.StdioServerTransport
import io.modelcontextprotocol.kotlin.sdk.types.Implementation
import io.modelcontextprotocol.kotlin.test.utils.Retry
import io.modelcontextprotocol.kotlin.test.utils.TypeScriptRunner
import io.modelcontextprotocol.kotlin.test.utils.isWindows
import kotlinx.coroutines.withTimeout
import kotlinx.io.Sink
import kotlinx.io.Source
import kotlinx.io.asSink
import kotlinx.io.asSource
import kotlinx.io.buffered
import org.awaitility.kotlin.await
import org.junit.jupiter.api.BeforeAll
import java.io.BufferedReader
import java.io.File
import java.io.InputStreamReader
import java.net.ServerSocket
import java.net.Socket
import java.util.concurrent.TimeUnit
import kotlin.io.path.createTempDirectory
import kotlin.time.Duration.Companion.seconds
enum class TransportKind { SSE, STDIO, DEFAULT }
enum class TransportKind { SSE, STDIO }
@Retry(times = 3)
abstract class TsTestBase {
protected open val transportKind: TransportKind = TransportKind.DEFAULT
protected abstract val transportKind: TransportKind
protected val projectRoot: File get() = File(System.getProperty("user.dir"))
protected val tsClientDir: File
get() {
val base = File(
companion object {
@JvmStatic
protected val projectRoot: File get() = File(System.getProperty("user.dir"))
@JvmStatic
protected val tsClientDir: File
get() = File(projectRoot, "src/jvmTest/typescript")
@JvmStatic
protected val tsSourceDir: File
get() = File(
projectRoot,
"src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript",
)
// Allow override via system property for CI: -Dts.transport=stdio|sse
val fromProp = System.getProperty("ts.transport")?.lowercase()
val overrideSubDir = when (fromProp) {
"stdio" -> "stdio"
"sse" -> "sse"
else -> null
}
@JvmStatic
private var sharedSseServer: Process? = null
val subDirName = overrideSubDir ?: when (transportKind) {
TransportKind.STDIO -> "stdio"
TransportKind.SSE -> "sse"
TransportKind.DEFAULT -> null
@JvmStatic
private var sharedSsePort: Int = 0
@JvmStatic
@Synchronized
protected fun getSharedSseUrl(): String {
if (sharedSseServer == null || !sharedSseServer!!.isAlive) {
sharedSsePort = io.modelcontextprotocol.kotlin.test.utils.findFreePort()
val server = TypeScriptServer(tsClientDir)
sharedSseServer = server.startSse(sharedSsePort)
println("Shared TypeScript SSE server started on port $sharedSsePort")
Runtime.getRuntime().addShutdownHook(
Thread {
sharedSseServer?.let {
println("Stopping shared TypeScript SSE server")
io.modelcontextprotocol.kotlin.test.utils.stopProcess(it)
}
},
)
}
if (subDirName != null) {
val sub = File(base, subDirName)
if (sub.exists()) return sub
}
return base
return "http://localhost:$sharedSsePort/mcp"
}
companion object {
@JvmStatic
private val tempRootDir: File = createTempDirectory("typescript-sdk-").toFile().apply { deleteOnExit() }
@ -72,6 +84,7 @@ abstract class TsTestBase {
@JvmStatic
@BeforeAll
@Deprecated("Use setupTypeScriptSdkWithDependencies instead")
fun setupTypeScriptSdk() {
println("Cloning TypeScript SDK repository")
@ -87,32 +100,43 @@ abstract class TsTestBase {
.redirectErrorStream(true)
.start()
val exitCode = process.waitFor()
if (exitCode != 0) {
throw RuntimeException("Failed to clone TypeScript SDK repository: exit code $exitCode")
require(exitCode == 0) {
"Failed to clone TypeScript SDK repository: exit code $exitCode"
}
}
println("Installing TypeScript SDK dependencies")
executeCommand("npm install", sdkDir, allowFailure = false, timeoutSeconds = null)
val npmInstallResult = executeCommand("npm install", sdkDir, allowFailure = true, timeoutSeconds = null)
if (npmInstallResult.contains("EUNSUPPORTEDPROTOCOL") || npmInstallResult.contains("catalog:")) {
println("npm install failed due to catalog protocol. Trying pnpm...")
try {
// Try to use npx pnpm to ensure it's available
executeCommand("npx pnpm install", sdkDir, allowFailure = false, timeoutSeconds = null)
} catch (e: Exception) {
println("pnpm install failed or not found: ${e.message}")
println("Attempting to patch package.json to remove catalog: protocol and use npm")
patchPackageJson(sdkDir)
// Also patch other package.json files in the workspace if they exist
sdkDir.walk().filter { it.name == "package.json" }.forEach { patchPackageJson(it.parentFile) }
executeCommand("npm install", sdkDir, allowFailure = false, timeoutSeconds = null)
}
}
}
@JvmStatic
protected fun killProcessOnPort(port: Int) {
val isWindows = System.getProperty("os.name").lowercase().contains("windows")
val killCommand = if (isWindows) {
"netstat -ano | findstr :$port | for /f \"tokens=5\" %a in ('more')" +
" do taskkill /F /PID %a 2>nul || echo No process found"
} else {
"lsof -ti:$port | xargs kill -9 2>/dev/null || true"
}
executeCommand(killCommand, File("."), allowFailure = true, timeoutSeconds = null)
}
@JvmStatic
protected fun findFreePort(): Int {
ServerSocket(0).use { socket ->
return socket.localPort
}
@Deprecated("never patch package.json")
private fun patchPackageJson(dir: File) {
val packageJson = File(dir, "package.json")
if (!packageJson.exists()) return
var content = packageJson.readText()
// Replace "catalog:something" with "*" or a reasonable default if we knew it.
// Since we don't know the exact versions in the catalog, "*" is a risky but possible fallback for tests.
// However, a better approach is to use pnpm if available.
content = content.replace(Regex(""""catalog:[^"]*""""), "\"*\"")
// Also replace workspace: protocol if we are using npm fallback
content = content.replace(Regex(""""workspace:[^"]*""""), "\"*\"")
// And any other unsupported protocols that might be in the package.json
content = content.replace(Regex(""""link:[^"]*""""), "\"*\"")
packageJson.writeText(content)
}
@JvmStatic
@ -132,7 +156,6 @@ abstract class TsTestBase {
throw RuntimeException("Working directory is not accessible: ${workingDir.absolutePath}")
}
val isWindows = System.getProperty("os.name").lowercase().contains("windows")
val processBuilder = if (isWindows) {
ProcessBuilder()
.command("cmd.exe", "/c", "set TYPESCRIPT_SDK_DIR=${sdkDir.absolutePath} && $command")
@ -157,11 +180,9 @@ abstract class TsTestBase {
if (timeoutSeconds == null) {
val exitCode = process.waitFor()
if (!allowFailure && exitCode != 0) {
throw RuntimeException(
"Command execution failed with exit code $exitCode: $command\n" +
"Working dir: ${workingDir.absolutePath}\nOutput:\n$output",
)
require(allowFailure || exitCode == 0) {
"Command execution failed with exit code $exitCode: $command\n" +
"Working dir: ${workingDir.absolutePath}\nOutput:\n$output"
}
} else {
process.waitFor(timeoutSeconds, TimeUnit.SECONDS)
@ -172,114 +193,24 @@ abstract class TsTestBase {
}
private fun waitForProcessTermination(process: Process, timeoutSeconds: Long): Boolean {
if (process.isAlive && !process.waitFor(timeoutSeconds, TimeUnit.SECONDS)) {
process.destroyForcibly()
process.waitFor(2, TimeUnit.SECONDS)
return false
}
io.modelcontextprotocol.kotlin.test.utils.stopProcess(process)
return true
}
private fun createProcessOutputReader(process: Process, prefix: String = "TS-SERVER"): Thread {
val outputReader = Thread {
try {
process.inputStream.bufferedReader().useLines { lines ->
for (line in lines) {
println("[$prefix] $line")
}
}
} catch (e: Exception) {
println("Warning: Error reading process output: ${e.message}")
}
}
outputReader.isDaemon = true
return outputReader
}
private fun createProcessErrorReader(process: Process, prefix: String = "TS-SERVER"): Thread {
val errorReader = Thread {
try {
process.errorStream.bufferedReader().useLines { lines ->
for (line in lines) {
println("[$prefix][err] $line")
}
}
} catch (e: Exception) {
println("Warning: Error reading process error stream: ${e.message}")
}
}
errorReader.isDaemon = true
return errorReader
}
protected fun waitForPort(host: String = "localhost", port: Int, timeoutSeconds: Long = 10): Boolean = try {
await.atMost(timeoutSeconds, TimeUnit.SECONDS)
.pollDelay(200, TimeUnit.MILLISECONDS)
.pollInterval(100, TimeUnit.MILLISECONDS)
.until {
try {
Socket(host, port).use { true }
} catch (_: Exception) {
false
}
}
true
} catch (_: Exception) {
false
}
protected fun waitForPort(host: String = "localhost", port: Int, timeoutSeconds: Long = 10): Boolean =
io.modelcontextprotocol.kotlin.test.utils.waitForPort(host, port, timeoutSeconds)
protected fun executeCommandAllowingFailure(command: String, workingDir: File, timeoutSeconds: Long = 20): String =
executeCommand(command, workingDir, allowFailure = true, timeoutSeconds = timeoutSeconds)
protected fun startTypeScriptServer(port: Int): Process {
killProcessOnPort(port)
if (!sdkDir.exists() || !sdkDir.isDirectory) {
throw IllegalStateException(
"TypeScript SDK directory does not exist or is not accessible: ${sdkDir.absolutePath}",
)
}
val isWindows = System.getProperty("os.name").lowercase().contains("windows")
val localServerPath = File(tsClientDir, "simpleStreamableHttp.ts").absolutePath
val processBuilder = if (isWindows) {
ProcessBuilder()
.command(
"cmd.exe",
"/c",
"set MCP_PORT=$port && set NODE_PATH=${sdkDir.absolutePath}\\node_modules && npx --prefix \"${sdkDir.absolutePath}\" tsx \"$localServerPath\"",
)
} else {
ProcessBuilder()
.command(
"bash",
"-c",
"MCP_PORT=$port NODE_PATH='${sdkDir.absolutePath}/node_modules' npx --prefix '${sdkDir.absolutePath}' tsx \"$localServerPath\"",
)
}
processBuilder.environment()["TYPESCRIPT_SDK_DIR"] = sdkDir.absolutePath
val process = processBuilder
.directory(tsClientDir)
.redirectErrorStream(true)
.start()
createProcessOutputReader(process).start()
if (!waitForPort(port = port, timeoutSeconds = 20)) {
throw IllegalStateException("TypeScript server did not become ready on localhost:$port within timeout")
}
return process
val server = TypeScriptServer(tsClientDir)
return server.startSse(port)
}
protected fun stopProcess(process: Process, waitSeconds: Long = 3, name: String = "TypeScript server") {
process.destroy()
if (waitForProcessTermination(process, waitSeconds)) {
println("$name stopped gracefully")
} else {
println("$name did not stop gracefully, forced termination")
}
io.modelcontextprotocol.kotlin.test.utils.stopProcess(process)
println("$name stopped")
}
// ===== SSE client helpers =====
@ -301,41 +232,8 @@ abstract class TsTestBase {
// ===== STDIO client + server helpers =====
protected fun startTypeScriptServerStdio(): Process {
if (!sdkDir.exists() || !sdkDir.isDirectory) {
throw IllegalStateException(
"TypeScript SDK directory does not exist or is not accessible: ${sdkDir.absolutePath}",
)
}
val isWindows = System.getProperty("os.name").lowercase().contains("windows")
val localServerPath = File(tsClientDir, "simpleStdio.ts").absolutePath
val processBuilder = if (isWindows) {
ProcessBuilder()
.command(
"cmd.exe",
"/c",
"set NODE_PATH=${sdkDir.absolutePath}\\node_modules && npx --prefix \"${sdkDir.absolutePath}\" tsx \"$localServerPath\"",
)
} else {
ProcessBuilder()
.command(
"bash",
"-c",
"NODE_PATH='${sdkDir.absolutePath}/node_modules' npx --prefix '${sdkDir.absolutePath}' tsx \"$localServerPath\"",
)
}
processBuilder.environment()["TYPESCRIPT_SDK_DIR"] = sdkDir.absolutePath
val process = processBuilder
.directory(tsClientDir)
.redirectErrorStream(false)
.start()
// For stdio transports, do NOT read from stdout (it's used for protocol). Read stderr for logs only.
createProcessErrorReader(process, prefix = "TS-SERVER-STDIO").start()
// Give the process a moment to start
await.atMost(2, TimeUnit.SECONDS)
.pollDelay(200, TimeUnit.MILLISECONDS)
.pollInterval(100, TimeUnit.MILLISECONDS)
.until { process.isAlive }
return process
val server = TypeScriptServer(tsClientDir)
return server.startStdio()
}
protected suspend fun newClientStdio(process: Process): Client {
@ -367,40 +265,20 @@ abstract class TsTestBase {
// ===== Helpers to run TypeScript client over STDIO against Kotlin server over STDIO =====
protected fun runStdioClient(vararg args: String): String {
// Start Node stdio client (it will speak MCP over its stdout/stdin)
val isWindows = System.getProperty("os.name").lowercase().contains("windows")
val clientPath = File(tsClientDir, "myClient.ts").absolutePath
val process = if (isWindows) {
ProcessBuilder()
.command(
"cmd.exe",
"/c",
(
"set TYPESCRIPT_SDK_DIR=${sdkDir.absolutePath} && " +
"set NODE_PATH=${sdkDir.absolutePath}\\node_modules && " +
"npx --prefix \"${sdkDir.absolutePath}\" tsx \"$clientPath\" " +
args.joinToString(" ")
),
)
.directory(tsClientDir)
.redirectErrorStream(false)
.start()
} else {
ProcessBuilder()
.command(
"bash",
"-c",
(
"TYPESCRIPT_SDK_DIR='${sdkDir.absolutePath}' " +
"NODE_PATH='${sdkDir.absolutePath}/node_modules' " +
"npx --prefix '${sdkDir.absolutePath}' tsx \"$clientPath\" " +
args.joinToString(" ")
),
)
.directory(tsClientDir)
.redirectErrorStream(false)
.start()
val subDirName = when (transportKind) {
TransportKind.STDIO -> "stdio"
TransportKind.SSE -> "sse"
}
val clientPath = File(tsSourceDir, "$subDirName/myClient.ts").absolutePath
val process = TypeScriptRunner.run(
typescriptDir = tsClientDir,
scriptPath = clientPath,
arguments = args.toList(),
env = mapOf("TYPESCRIPT_SDK_DIR" to sdkDir.absolutePath),
redirectErrorStream = false,
logPrefix = "TS-CLIENT-STDIO",
)
// Create Kotlin server and attach stdio transport to the process streams
val server: Server = KotlinServerForTsClient().createMcpServer()

View file

@ -0,0 +1,121 @@
package io.modelcontextprotocol.kotlin.sdk.integration.typescript
import io.modelcontextprotocol.kotlin.test.utils.TypeScriptRunner
import io.modelcontextprotocol.kotlin.test.utils.killProcessOnPort
import io.modelcontextprotocol.kotlin.test.utils.stopProcess
import io.modelcontextprotocol.kotlin.test.utils.waitForPort
import org.awaitility.kotlin.await
import java.io.File
import java.util.concurrent.TimeUnit
/**
* Factory class for starting and managing TypeScript MCP servers for integration tests.
*
* This class provides methods to:
* - Start TypeScript servers using SSE or STDIO transports via npx tsx
* - Install npm dependencies in the TypeScript test directory
* - Manage server lifecycle (start/stop)
* - Handle port management and process cleanup
*
* The servers are started using npx tsx to execute TypeScript files directly:
* - `npx tsx server/stdio-server.ts` for STDIO transport
* - `npx tsx server/sse-server.ts` for SSE transport
*
* Usage:
* ```kotlin
* val typescriptDir = File("kotlin-sdk-test/src/jvmTest/typescript")
* val server = TypeScriptServer(typescriptDir)
*
* // Start SSE server
* val port = TypeScriptServer.findFreePort()
* val process = server.startSse(port)
*
* // Stop server
* server.stop()
* ```
*/
class TypeScriptServer(private val typescriptDir: File) {
private var process: Process? = null
/**
* Starts a TypeScript MCP server using SSE transport.
*
* @param port The port number to bind the SSE server to
* @return The started Process
* @throws IllegalStateException if the server fails to start within timeout
*/
fun startSse(port: Int): Process {
killProcessOnPort(port)
val serverPath = File(typescriptDir, "server/sse-server.ts").absolutePath
val proc = TypeScriptRunner.run(
typescriptDir = typescriptDir,
scriptPath = serverPath,
env = mapOf("MCP_PORT" to port.toString()),
redirectErrorStream = true,
logPrefix = "TS-SERVER-SSE",
)
process = proc
check(waitForPort(port = port, timeoutSeconds = 10)) {
proc.destroyForcibly()
"TypeScript SSE server did not become ready on localhost:$port within timeout"
}
return proc
}
/**
* Starts a TypeScript MCP server using STDIO transport.
*
* @return The started Process
*/
fun startStdio(): Process {
val serverPath = File(typescriptDir, "server/stdio-server.ts").absolutePath
val proc = TypeScriptRunner.run(
typescriptDir = typescriptDir,
scriptPath = serverPath,
redirectErrorStream = false,
logPrefix = "TS-SERVER-STDIO",
)
process = proc
await.atMost(5, TimeUnit.SECONDS)
.pollDelay(200, TimeUnit.MILLISECONDS)
.pollInterval(100, TimeUnit.MILLISECONDS)
.until { proc.isAlive }
return proc
}
/**
* Stops the currently running server process.
*
* Attempts graceful shutdown first, then forces termination if necessary.
*/
fun stop() {
process?.let {
stopProcess(it)
}
process = null
}
companion object {
/**
* Installs npm dependencies in the specified TypeScript directory.
*
* @param typescriptDir The directory containing package.json
* @throws RuntimeException if npm install fails
*/
@JvmStatic
fun installDependencies(typescriptDir: File) {
TypeScriptRunner.installDependencies(typescriptDir)
}
/**
* Finds an available port by opening and closing a ServerSocket.
*
* @return An available port number
*/
@JvmStatic
fun findFreePort(): Int = io.modelcontextprotocol.kotlin.test.utils.findFreePort()
}
}

View file

@ -4,7 +4,6 @@ import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TsTestBase
import io.modelcontextprotocol.kotlin.sdk.types.TextContent
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
@ -14,7 +13,6 @@ import kotlinx.coroutines.withTimeout
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.Timeout
import java.util.concurrent.TimeUnit
@ -23,25 +21,14 @@ import kotlin.test.assertNotNull
import kotlin.test.assertTrue
import kotlin.time.Duration.Companion.seconds
@DisabledOnCI
class KotlinClientTsServerEdgeCasesTestSse : TsTestBase() {
override val transportKind = TransportKind.SSE
private var port: Int = 0
private val host = "localhost"
private lateinit var serverUrl: String
private val serverUrl: String by lazy { getSharedSseUrl() }
private lateinit var client: Client
private lateinit var tsServerProcess: Process
@BeforeEach
fun setUp() {
port = findFreePort()
serverUrl = "http://$host:$port/mcp"
tsServerProcess = startTypeScriptServer(port)
println("TypeScript server started on port $port")
}
@AfterEach
fun tearDown() {
@ -56,15 +43,6 @@ class KotlinClientTsServerEdgeCasesTestSse : TsTestBase() {
println("Warning: Error during client close: ${e.message}")
}
}
if (::tsServerProcess.isInitialized) {
try {
println("Stopping TypeScript server")
stopProcess(tsServerProcess)
} catch (e: Exception) {
println("Warning: Error during TypeScript server stop: ${e.message}")
}
}
}
@Test

View file

@ -3,41 +3,14 @@ package io.modelcontextprotocol.kotlin.sdk.integration.typescript.sse
import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.AbstractKotlinClientTsServerTest
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
import kotlinx.coroutines.withTimeout
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach
import kotlin.time.Duration.Companion.seconds
@DisabledOnCI
class KotlinClientTsServerTestSse : AbstractKotlinClientTsServerTest() {
override val transportKind = TransportKind.SSE
private var port: Int = 0
private val host = "localhost"
private lateinit var serverUrl: String
private lateinit var tsServerProcess: Process
@BeforeEach
fun setUpSse() {
port = findFreePort()
serverUrl = "http://$host:$port/mcp"
tsServerProcess = startTypeScriptServer(port)
println("TypeScript server started on port $port")
}
@AfterEach
fun tearDownSse() {
if (::tsServerProcess.isInitialized) {
try {
println("Stopping TypeScript server")
stopProcess(tsServerProcess)
} catch (e: Exception) {
println("Warning: Error during TypeScript server stop: ${e.message}")
}
}
}
private val serverUrl: String by lazy { getSharedSseUrl() }
override suspend fun <T> useClient(block: suspend (Client) -> T): T = withClient(serverUrl) { client ->
try {

View file

@ -4,7 +4,6 @@ import io.modelcontextprotocol.kotlin.sdk.TextContent
import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TsTestBase
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
@ -14,7 +13,6 @@ import kotlinx.coroutines.withTimeout
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.Timeout
import java.util.concurrent.TimeUnit
@ -23,25 +21,14 @@ import kotlin.test.assertNotNull
import kotlin.test.assertTrue
import kotlin.time.Duration.Companion.seconds
@DisabledOnCI
class OldSchemaKotlinClientTsServerEdgeCasesTestSse : TsTestBase() {
override val transportKind = TransportKind.SSE
private var port: Int = 0
private val host = "localhost"
private lateinit var serverUrl: String
private val serverUrl: String by lazy { getSharedSseUrl() }
private lateinit var client: Client
private lateinit var tsServerProcess: Process
@BeforeEach
fun setUp() {
port = findFreePort()
serverUrl = "http://$host:$port/mcp"
tsServerProcess = startTypeScriptServer(port)
println("TypeScript server started on port $port")
}
@AfterEach
fun tearDown() {
@ -56,15 +43,6 @@ class OldSchemaKotlinClientTsServerEdgeCasesTestSse : TsTestBase() {
println("Warning: Error during client close: ${e.message}")
}
}
if (::tsServerProcess.isInitialized) {
try {
println("Stopping TypeScript server")
stopProcess(tsServerProcess)
} catch (e: Exception) {
println("Warning: Error during TypeScript server stop: ${e.message}")
}
}
}
@Test

View file

@ -3,41 +3,15 @@ package io.modelcontextprotocol.kotlin.sdk.integration.typescript.sse
import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.OldSchemaAbstractKotlinClientTsServerTest
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
import kotlinx.coroutines.withTimeout
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach
import kotlin.time.Duration.Companion.seconds
@DisabledOnCI
class OldSchemaKotlinClientTsServerTestSse : OldSchemaAbstractKotlinClientTsServerTest() {
override val transportKind = TransportKind.SSE
private var port: Int = 0
private val host = "localhost"
private lateinit var serverUrl: String
private lateinit var tsServerProcess: Process
@BeforeEach
fun setUpSse() {
port = findFreePort()
serverUrl = "http://$host:$port/mcp"
tsServerProcess = startTypeScriptServer(port)
println("TypeScript server started on port $port")
}
@AfterEach
fun tearDownSse() {
if (::tsServerProcess.isInitialized) {
try {
println("Stopping TypeScript server")
stopProcess(tsServerProcess)
} catch (e: Exception) {
println("Warning: Error during TypeScript server stop: ${e.message}")
}
}
}
private val serverUrl: String by lazy { getSharedSseUrl() }
override suspend fun <T> useClient(block: suspend (Client) -> T): T = withClient(serverUrl) { client ->
try {

View file

@ -3,6 +3,8 @@ package io.modelcontextprotocol.kotlin.sdk.integration.typescript.sse
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.AbstractTsClientKotlinServerTest
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
import io.modelcontextprotocol.kotlin.test.utils.findFreePort
import io.modelcontextprotocol.kotlin.test.utils.killProcessOnPort
import org.junit.jupiter.api.AfterEach
import org.junit.jupiter.api.BeforeEach

View file

@ -3,6 +3,8 @@ package io.modelcontextprotocol.kotlin.sdk.integration.typescript.sse
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TsTestBase
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
import io.modelcontextprotocol.kotlin.test.utils.findFreePort
import io.modelcontextprotocol.kotlin.test.utils.killProcessOnPort
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
@ -35,8 +37,8 @@ class TsEdgeCasesTestSse : TsTestBase() {
killProcessOnPort(port)
httpServer = KotlinServerForTsClient()
httpServer?.start(port)
if (!waitForPort(port = port)) {
throw IllegalStateException("Kotlin test server did not become ready on localhost:$port within timeout")
check(waitForPort(port = port)) {
"Kotlin test server did not become ready on localhost:$port within timeout"
}
println("Kotlin server started on port $port")
}

View file

@ -17,8 +17,8 @@ async function main() {
const path = await import('path');
// @ts-ignore
const { pathToFileURL } = await import('url');
const clientUrl = pathToFileURL(path.join(sdkDir, 'src', 'client', 'index.ts')).href;
const streamUrl = pathToFileURL(path.join(sdkDir, 'src', 'client', 'streamableHttp.js')).href;
const clientUrl = pathToFileURL(path.join(sdkDir, 'packages', 'client', 'src', 'index.ts')).href;
const streamUrl = pathToFileURL(path.join(sdkDir, 'packages', 'client', 'src', 'streamableHttp.ts')).href;
// @ts-ignore
({ Client } = await import(clientUrl));
// @ts-ignore

View file

@ -1,701 +0,0 @@
// @ts-nocheck
import express, {Request, Response} from 'express';
import {randomUUID} from 'node:crypto';
import {z} from 'zod';
import cors from 'cors';
import path from 'node:path';
import {pathToFileURL} from 'node:url';
const SDK_DIR = process.env.TYPESCRIPT_SDK_DIR;
if (!SDK_DIR) {
throw new Error('TYPESCRIPT_SDK_DIR environment variable is not set. It should point to the cloned typescript-sdk directory.');
}
async function importFromSdk<T = any>(rel: string): Promise<T> {
const full = path.resolve(SDK_DIR!, rel);
const url = pathToFileURL(full).href;
return (await import(url)) as unknown as T;
}
async function main() {
const {McpServer} = await importFromSdk<any>('src/server/mcp.ts');
const {StreamableHTTPServerTransport} = await importFromSdk<any>('src/server/streamableHttp.ts');
const {
getOAuthProtectedResourceMetadataUrl,
mcpAuthMetadataRouter
} = await importFromSdk<any>('src/server/auth/router.ts');
const {requireBearerAuth} = await importFromSdk<any>('src/server/auth/middleware/bearerAuth.ts');
const {
isInitializeRequest,
} = await importFromSdk<any>('src/types.ts');
const {InMemoryEventStore} = await importFromSdk<any>('src/examples/shared/inMemoryEventStore.ts');
const {setupAuthServer} = await importFromSdk<any>('src/examples/server/demoInMemoryOAuthProvider.ts');
const {checkResourceAllowed} = await importFromSdk<any>('src/shared/auth-utils.ts');
// Check for OAuth flag
const useOAuth = process.argv.includes('--oauth');
const strictOAuth = process.argv.includes('--oauth-strict');
// Create an MCP server with implementation details
const getServer = () => {
const server = new McpServer({
name: 'simple-streamable-http-server',
version: '1.0.0'
}, {capabilities: {logging: {}}});
// Register a simple tool that returns a greeting
server.registerTool(
'greet',
{
title: 'Greeting Tool', // Display name for UI
description: 'A simple greeting tool',
inputSchema: {
name: z.string().describe('Name to greet'),
},
},
async ({name}): Promise<CallToolResult> => {
return {
content: [
{
type: 'text',
text: `Hello, ${name}!`,
},
],
};
}
);
// Register a tool that sends multiple greetings with notifications (with annotations)
server.tool(
'multi-greet',
'A tool that sends different greetings with delays between them',
{
name: z.string().describe('Name to greet'),
},
{
title: 'Multiple Greeting Tool',
readOnlyHint: true,
openWorldHint: false
},
async ({name}, extra): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
await server.sendLoggingMessage({
level: "debug",
data: `Starting multi-greet for ${name}`
}, extra.sessionId);
await sleep(1000); // Wait 1 second before first greeting
await server.sendLoggingMessage({
level: "info",
data: `Sending first greeting to ${name}`
}, extra.sessionId);
await sleep(1000); // Wait another second before second greeting
await server.sendLoggingMessage({
level: "info",
data: `Sending second greeting to ${name}`
}, extra.sessionId);
return {
content: [
{
type: 'text',
text: `Good morning, ${name}!`,
}
],
};
}
);
// Register a tool that demonstrates elicitation (user input collection)
// This creates a closure that captures the server instance
server.tool(
'collect-user-info',
'A tool that collects user information through elicitation',
{
infoType: z.enum(['contact', 'preferences', 'feedback']).describe('Type of information to collect'),
},
async ({infoType}): Promise<CallToolResult> => {
let message: string;
let requestedSchema: {
type: 'object';
properties: Record<string, PrimitiveSchemaDefinition>;
required?: string[];
};
switch (infoType) {
case 'contact':
message = 'Please provide your contact information';
requestedSchema = {
type: 'object',
properties: {
name: {
type: 'string',
title: 'Full Name',
description: 'Your full name',
},
email: {
type: 'string',
title: 'Email Address',
description: 'Your email address',
format: 'email',
},
phone: {
type: 'string',
title: 'Phone Number',
description: 'Your phone number (optional)',
},
},
required: ['name', 'email'],
};
break;
case 'preferences':
message = 'Please set your preferences';
requestedSchema = {
type: 'object',
properties: {
theme: {
type: 'string',
title: 'Theme',
description: 'Choose your preferred theme',
enum: ['light', 'dark', 'auto'],
enumNames: ['Light', 'Dark', 'Auto'],
},
notifications: {
type: 'boolean',
title: 'Enable Notifications',
description: 'Would you like to receive notifications?',
default: true,
},
frequency: {
type: 'string',
title: 'Notification Frequency',
description: 'How often would you like notifications?',
enum: ['daily', 'weekly', 'monthly'],
enumNames: ['Daily', 'Weekly', 'Monthly'],
},
},
required: ['theme'],
};
break;
case 'feedback':
message = 'Please provide your feedback';
requestedSchema = {
type: 'object',
properties: {
rating: {
type: 'integer',
title: 'Rating',
description: 'Rate your experience (1-5)',
minimum: 1,
maximum: 5,
},
comments: {
type: 'string',
title: 'Comments',
description: 'Additional comments (optional)',
maxLength: 500,
},
recommend: {
type: 'boolean',
title: 'Would you recommend this?',
description: 'Would you recommend this to others?',
},
},
required: ['rating', 'recommend'],
};
break;
default:
throw new Error(`Unknown info type: ${infoType}`);
}
try {
// Use the underlying server instance to elicit input from the client
const result = await server.server.elicitInput({
message,
requestedSchema,
});
if (result.action === 'accept') {
return {
content: [
{
type: 'text',
text: `Thank you! Collected ${infoType} information: ${JSON.stringify(result.content, null, 2)}`,
},
],
};
} else if (result.action === 'decline') {
return {
content: [
{
type: 'text',
text: `No information was collected. User declined ${infoType} information request.`,
},
],
};
} else {
return {
content: [
{
type: 'text',
text: `Information collection was cancelled by the user.`,
},
],
};
}
} catch (error) {
return {
content: [
{
type: 'text',
text: `Error collecting ${infoType} information: ${error}`,
},
],
};
}
}
);
// Register a simple prompt with title
server.registerPrompt(
'greeting-template',
{
title: 'Greeting Template', // Display name for UI
description: 'A simple greeting prompt template',
argsSchema: {
name: z.string().describe('Name to include in greeting'),
},
},
async ({name}): Promise<GetPromptResult> => {
return {
messages: [
{
role: 'user',
content: {
type: 'text',
text: `Please greet ${name} in a friendly manner.`,
},
},
],
};
}
);
// Register a tool specifically for testing resumability
server.tool(
'start-notification-stream',
'Starts sending periodic notifications for testing resumability',
{
interval: z.number().describe('Interval in milliseconds between notifications').default(100),
count: z.number().describe('Number of notifications to send (0 for 100)').default(50),
},
async ({interval, count}, extra): Promise<CallToolResult> => {
const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms));
let counter = 0;
while (count === 0 || counter < count) {
counter++;
try {
await server.sendLoggingMessage({
level: "info",
data: `Periodic notification #${counter} at ${new Date().toISOString()}`
}, extra.sessionId);
} catch (error) {
console.error("Error sending notification:", error);
}
// Wait for the specified interval
await sleep(interval);
}
return {
content: [
{
type: 'text',
text: `Started sending periodic notifications every ${interval}ms`,
}
],
};
}
);
// Create a simple resource at a fixed URI
server.registerResource(
'greeting-resource',
'https://example.com/greetings/default',
{
title: 'Default Greeting', // Display name for UI
description: 'A simple greeting resource',
mimeType: 'text/plain'
},
async (): Promise<ReadResourceResult> => {
return {
contents: [
{
uri: 'https://example.com/greetings/default',
text: 'Hello, world!',
},
],
};
}
);
// Create additional resources for ResourceLink demonstration
server.registerResource(
'example-file-1',
'file:///example/file1.txt',
{
title: 'Example File 1',
description: 'First example file for ResourceLink demonstration',
mimeType: 'text/plain'
},
async (): Promise<ReadResourceResult> => {
return {
contents: [
{
uri: 'file:///example/file1.txt',
text: 'This is the content of file 1',
},
],
};
}
);
server.registerResource(
'example-file-2',
'file:///example/file2.txt',
{
title: 'Example File 2',
description: 'Second example file for ResourceLink demonstration',
mimeType: 'text/plain'
},
async (): Promise<ReadResourceResult> => {
return {
contents: [
{
uri: 'file:///example/file2.txt',
text: 'This is the content of file 2',
},
],
};
}
);
// Register a tool that returns ResourceLinks
server.registerTool(
'list-files',
{
title: 'List Files with ResourceLinks',
description: 'Returns a list of files as ResourceLinks without embedding their content',
inputSchema: {
includeDescriptions: z.boolean().optional().describe('Whether to include descriptions in the resource links'),
},
},
async ({includeDescriptions = true}): Promise<CallToolResult> => {
const resourceLinks: ResourceLink[] = [
{
type: 'resource_link',
uri: 'https://example.com/greetings/default',
name: 'Default Greeting',
mimeType: 'text/plain',
...(includeDescriptions && {description: 'A simple greeting resource'})
},
{
type: 'resource_link',
uri: 'file:///example/file1.txt',
name: 'Example File 1',
mimeType: 'text/plain',
...(includeDescriptions && {description: 'First example file for ResourceLink demonstration'})
},
{
type: 'resource_link',
uri: 'file:///example/file2.txt',
name: 'Example File 2',
mimeType: 'text/plain',
...(includeDescriptions && {description: 'Second example file for ResourceLink demonstration'})
}
];
return {
content: [
{
type: 'text',
text: 'Here are the available files as resource links:',
},
...resourceLinks,
{
type: 'text',
text: '\nYou can read any of these resources using their URI.',
}
],
};
}
);
return server;
};
const MCP_PORT = process.env.MCP_PORT ? parseInt(process.env.MCP_PORT, 10) : 3000;
const AUTH_PORT = process.env.MCP_AUTH_PORT ? parseInt(process.env.MCP_AUTH_PORT, 10) : 3001;
const app = express();
app.use(express.json());
// Allow CORS all domains, expose the Mcp-Session-Id header
app.use(cors({
origin: '*', // Allow all origins
exposedHeaders: ["Mcp-Session-Id"]
}));
// Set up OAuth if enabled
let authMiddleware = null;
if (useOAuth) {
// Create auth middleware for MCP endpoints
const mcpServerUrl = new URL(`http://localhost:${MCP_PORT}/mcp`);
const authServerUrl = new URL(`http://localhost:${AUTH_PORT}`);
const oauthMetadata: OAuthMetadata = setupAuthServer({
authServerUrl,
mcpServerUrl,
strictResource: strictOAuth
});
const tokenVerifier = {
verifyAccessToken: async (token: string) => {
const endpoint = oauthMetadata.introspection_endpoint;
if (!endpoint) {
throw new Error('No token verification endpoint available in metadata');
}
const response = await fetch(endpoint, {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
},
body: new URLSearchParams({
token: token
}).toString()
});
if (!response.ok) {
throw new Error(`Invalid or expired token: ${await response.text()}`);
}
const data = await response.json();
if (strictOAuth) {
if (!data.aud) {
throw new Error(`Resource Indicator (RFC8707) missing`);
}
if (!checkResourceAllowed({requestedResource: data.aud, configuredResource: mcpServerUrl})) {
throw new Error(`Expected resource indicator ${mcpServerUrl}, got: ${data.aud}`);
}
}
// Convert the response to AuthInfo format
return {
token,
clientId: data.client_id,
scopes: data.scope ? data.scope.split(' ') : [],
expiresAt: data.exp,
};
}
}
// Add metadata routes to the main MCP server
app.use(mcpAuthMetadataRouter({
oauthMetadata,
resourceServerUrl: mcpServerUrl,
scopesSupported: ['mcp:tools'],
resourceName: 'MCP Demo Server',
}));
authMiddleware = requireBearerAuth({
verifier: tokenVerifier,
requiredScopes: [],
resourceMetadataUrl: getOAuthProtectedResourceMetadataUrl(mcpServerUrl),
});
}
// Map to store transports by session ID
const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {};
// MCP POST endpoint with optional auth
const mcpPostHandler = async (req: Request, res: Response) => {
const sessionId = req.headers['mcp-session-id'] as string | undefined;
if (sessionId) {
console.log(`Received MCP request for session: ${sessionId}`);
} else {
console.log('Request body:', req.body);
}
if (useOAuth && req.auth) {
console.log('Authenticated user:', req.auth);
}
try {
let transport: StreamableHTTPServerTransport;
if (sessionId && transports[sessionId]) {
// Reuse existing transport
transport = transports[sessionId];
} else if (!sessionId && isInitializeRequest(req.body)) {
// New initialization request
const eventStore = new InMemoryEventStore();
transport = new StreamableHTTPServerTransport({
sessionIdGenerator: () => randomUUID(),
eventStore, // Enable resumability
onsessioninitialized: (sessionId) => {
// Store the transport by session ID when session is initialized
// This avoids race conditions where requests might come in before the session is stored
console.log(`Session initialized with ID: ${sessionId}`);
transports[sessionId] = transport;
}
});
// Set up onclose handler to clean up transport when closed
transport.onclose = () => {
const sid = transport.sessionId;
if (sid && transports[sid]) {
console.log(`Transport closed for session ${sid}, removing from transports map`);
delete transports[sid];
}
};
// Connect the transport to the MCP server BEFORE handling the request
// so responses can flow back through the same transport
const server = getServer();
await server.connect(transport);
await transport.handleRequest(req, res, req.body);
return; // Already handled
} else {
// Invalid request - no session ID or not initialization request
res.status(400).json({
jsonrpc: '2.0',
error: {
code: -32000,
message: 'Bad Request: No valid session ID provided',
},
id: null,
});
return;
}
// Handle the request with existing transport - no need to reconnect
// The existing transport is already connected to the server
await transport.handleRequest(req, res, req.body);
} catch (error) {
console.error('Error handling MCP request:', error);
if (!res.headersSent) {
res.status(500).json({
jsonrpc: '2.0',
error: {
code: -32603,
message: 'Internal server error',
},
id: null,
});
}
}
};
// Set up routes with conditional auth middleware
if (useOAuth && authMiddleware) {
app.post('/mcp', authMiddleware, mcpPostHandler);
} else {
app.post('/mcp', mcpPostHandler);
}
// Handle GET requests for SSE streams (using built-in support from StreamableHTTP)
const mcpGetHandler = async (req: Request, res: Response) => {
const sessionId = req.headers['mcp-session-id'] as string | undefined;
if (!sessionId || !transports[sessionId]) {
res.status(400).send('Invalid or missing session ID');
return;
}
if (useOAuth && req.auth) {
console.log('Authenticated SSE connection from user:', req.auth);
}
// Check for Last-Event-ID header for resumability
const lastEventId = req.headers['last-event-id'] as string | undefined;
if (lastEventId) {
console.log(`Client reconnecting with Last-Event-ID: ${lastEventId}`);
} else {
console.log(`Establishing new SSE stream for session ${sessionId}`);
}
const transport = transports[sessionId];
await transport.handleRequest(req, res);
};
// Set up GET route with conditional auth middleware
if (useOAuth && authMiddleware) {
app.get('/mcp', authMiddleware, mcpGetHandler);
} else {
app.get('/mcp', mcpGetHandler);
}
// Handle DELETE requests for session termination (according to MCP spec)
const mcpDeleteHandler = async (req: Request, res: Response) => {
const sessionId = req.headers['mcp-session-id'] as string | undefined;
if (!sessionId || !transports[sessionId]) {
res.status(400).send('Invalid or missing session ID');
return;
}
console.log(`Received session termination request for session ${sessionId}`);
try {
const transport = transports[sessionId];
await transport.handleRequest(req, res);
} catch (error) {
console.error('Error handling session termination:', error);
if (!res.headersSent) {
res.status(500).send('Error processing session termination');
}
}
};
// Set up DELETE route with conditional auth middleware
if (useOAuth && authMiddleware) {
app.delete('/mcp', authMiddleware, mcpDeleteHandler);
} else {
app.delete('/mcp', mcpDeleteHandler);
}
app.listen(MCP_PORT, (error) => {
if (error) {
console.error('Failed to start server:', error);
process.exit(1);
}
console.log(`MCP Streamable HTTP Server listening on port ${MCP_PORT}`);
});
// Handle server shutdown
process.on('SIGINT', async () => {
console.log('Shutting down server...');
// Close all active transports to properly clean up resources
for (const sessionId in transports) {
try {
console.log(`Closing transport for session ${sessionId}`);
await transports[sessionId].close();
delete transports[sessionId];
} catch (error) {
console.error(`Error closing transport for session ${sessionId}:`, error);
}
}
console.log('Server shutdown complete');
process.exit(0);
});
}
main().catch((err) => {
console.error('Failed to start server:', err);
process.exit(1);
});

View file

@ -4,7 +4,6 @@ import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TsTestBase
import io.modelcontextprotocol.kotlin.sdk.types.TextContent
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
@ -19,7 +18,6 @@ import kotlin.test.assertEquals
import kotlin.test.assertNotNull
import kotlin.test.assertTrue
@DisabledOnCI
class KotlinClientTsServerEdgeCasesTestStdio : TsTestBase() {
override val transportKind = TransportKind.STDIO

View file

@ -3,9 +3,7 @@ package io.modelcontextprotocol.kotlin.sdk.integration.typescript.stdio
import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.AbstractKotlinClientTsServerTest
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
@DisabledOnCI
class KotlinClientTsServerTestStdio : AbstractKotlinClientTsServerTest() {
override val transportKind = TransportKind.STDIO

View file

@ -4,7 +4,6 @@ import io.modelcontextprotocol.kotlin.sdk.TextContent
import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TsTestBase
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
@ -19,7 +18,6 @@ import kotlin.test.assertEquals
import kotlin.test.assertNotNull
import kotlin.test.assertTrue
@DisabledOnCI
class OldSchemaKotlinClientTsServerEdgeCasesTestStdio : TsTestBase() {
override val transportKind = TransportKind.STDIO

View file

@ -3,9 +3,7 @@ package io.modelcontextprotocol.kotlin.sdk.integration.typescript.stdio
import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.OldSchemaAbstractKotlinClientTsServerTest
import io.modelcontextprotocol.kotlin.sdk.integration.typescript.TransportKind
import io.modelcontextprotocol.kotlin.test.utils.DisabledOnCI
@DisabledOnCI
class OldSchemaKotlinClientTsServerTestStdio : OldSchemaAbstractKotlinClientTsServerTest() {
override val transportKind = TransportKind.STDIO

View file

@ -16,7 +16,7 @@ async function main() {
const path = await import('path');
const { pathToFileURL } = await import('url');
const clientUrl = pathToFileURL(path.join(sdkDir, 'src', 'client', 'index.ts')).href;
const clientUrl = pathToFileURL(path.join(sdkDir, 'packages', 'client', 'src', 'index.ts')).href;
({ Client } = await import(clientUrl));
if (!toolName) {

Some files were not shown because too many files have changed in this diff Show more