Custom Ktor Plugins

← Ktor Lightweight APIs

Goal

Build a set of custom Ktor plugins to deepen your understanding of the plugin system. Ktor plugins are the equivalent of Express.js middleware and Go middleware functions — they intercept the request/response pipeline to add cross-cutting behaviour like timing, tracing, rate limiting, and auth.

What you’ll build

Four plugins, each demonstrating a different slice of the plugin API, plus a demo server that installs all of them:

1. ResponseTimingPlugin — adds an X-Response-Time header to every response with the request duration in milliseconds. Demonstrates the onCall and onCallRespond lifecycle hooks and passing data between them via call.attributes. Equivalent to Express’s response-time package or Go timing middleware.
2. RequestIdPlugin — ensures every request has an X-Request-Id header, preserving a client-provided ID or generating a new UUID. Demonstrates configurable plugins via createConfiguration and reading request / writing response headers. Equivalent to Express’s express-request-id or Go request-ID middleware.
3. RateLimitPlugin — IP-based rate limiting with a configurable limit and window. Demonstrates stateful plugins with shared data structures, short-circuiting a request (calling call.respond before the handler runs), and standard rate-limit headers (X-RateLimit-Limit, X-RateLimit-Remaining, Retry-After). Equivalent to Express’s express-rate-limit or Go’s x/time/rate.
4. ApiKeyAuthPlugin — a custom API-key scheme that plugs into Ktor’s built-in Authentication plugin. Demonstrates a custom AuthenticationProvider with challenge handling and a Principal for reading authenticated client info in routes. Equivalent to custom auth middleware in Express or Go.

Plugin anatomy

Every Ktor plugin built with createApplicationPlugin follows the same shape: a name, an optional config factory, and a body that registers lifecycle hooks. The TS/Go middleware you already know maps cleanly onto those hooks.

val MyPlugin = createApplicationPlugin(
    name = "MyPlugin",
    createConfiguration = ::MyPluginConfig  // optional config
) {
    onCall { call ->              // before the route handler
        // read headers, set attributes, short-circuit with call.respond
    }
    onCallRespond { call, body -> // after the handler, before the response is sent
        // modify headers, transform the body
    }
    onCallReceive { call, body -> // when call.receive<T>() is invoked
        // transform or validate the request body
    }
}

// install it:
install(MyPlugin) {
    // configure via the config class
}

The mental model versus middleware in other stacks:

TypeScript

// Express: one function, you call next() to continue
app.use((req, res, next) => {
  const start = Date.now();
  res.on("finish", () => {
    res.setHeader("X-Response-Time", `${Date.now() - start}ms`);
  });
  next();
});

Go

// net/http: wrap the next handler
func ResponseTiming(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        start := time.Now()
        next.ServeHTTP(w, r)
        w.Header().Set("X-Response-Time", fmt.Sprintf("%dms", time.Since(start).Milliseconds()))
    })
}

Kotlin

// Ktor: register hooks instead of wrapping a handler
val ResponseTimingPlugin = createApplicationPlugin(name = "ResponseTiming") {
    onCall { call -> call.attributes.put(startKey, System.currentTimeMillis()) }
    onCallRespond { call, _ ->
        val start = call.attributes[startKey]
        call.response.headers.append("X-Response-Time", "${System.currentTimeMillis() - start}ms")
    }
}

Where Express and Go give you one linear function and you control flow with next() / next.ServeHTTP(...), Ktor splits the work across named hooks (onCall, onCallRespond, onCallReceive) and the framework drives the pipeline. Data that needs to survive between hooks rides on call.attributes rather than a closure variable.

The worked solution

A single Ktor application module installs all four plugins. The project layout:

• Directoryktor-plugins/

  • build.gradle.kts Ktor + serialization deps
  • settings.gradle.kts project name
  • Directorysrc/main/kotlin/com/example/plugins/

    • Application.kt entry point, installs all plugins, defines routes
    • ResponseTimingPlugin.kt X-Response-Time header
    • RequestIdPlugin.kt X-Request-Id header
    • RateLimitPlugin.kt IP-based rate limiting
    • ApiKeyAuthPlugin.kt custom API-key authentication
  • Directorysrc/test/kotlin/com/example/plugins/ one test per plugin

    • …

build.gradle.kts

The Ktor Gradle plugin pulls the framework together. The notable dependencies are ktor-server-core (the plugin API lives here), ktor-server-netty (the engine), ktor-server-auth (the built-in Authentication plugin our API-key scheme hooks into), and the kotlinx JSON serialization adapter.

build.gradle.kts

plugins {
    kotlin("jvm") version "2.1.0"
    kotlin("plugin.serialization") version "2.1.0"
    id("io.ktor.plugin") version "3.1.0"
}

group = "com.example"
version = "1.0.0"

application {
    mainClass.set("com.example.plugins.ApplicationKt")
}

java {
    toolchain {
        languageVersion = JavaLanguageVersion.of(21)
    }
}

repositories {
    mavenCentral()
}

val ktorVersion = "3.1.0"
val logbackVersion = "1.5.15"

dependencies {
    // Ktor server
    implementation("io.ktor:ktor-server-core:$ktorVersion")
    implementation("io.ktor:ktor-server-netty:$ktorVersion")
    implementation("io.ktor:ktor-server-content-negotiation:$ktorVersion")
    implementation("io.ktor:ktor-server-status-pages:$ktorVersion")
    implementation("io.ktor:ktor-server-auth:$ktorVersion")
    implementation("io.ktor:ktor-server-request-validation:$ktorVersion")

    // Serialization
    implementation("io.ktor:ktor-serialization-kotlinx-json:$ktorVersion")

    // Logging
    implementation("ch.qos.logback:logback-classic:$logbackVersion")

    // Testing
    testImplementation("io.ktor:ktor-server-test-host:$ktorVersion")
    testImplementation("io.ktor:ktor-client-content-negotiation:$ktorVersion")
    testImplementation("org.jetbrains.kotlin:kotlin-test-junit5")
    testRuntimeOnly("org.junit.platform:junit-platform-launcher")
}

tasks.withType<Test> {
    useJUnitPlatform()
}

ResponseTimingPlugin.kt — two hooks sharing an attribute

This is the simplest plugin and the clearest illustration of why hooks beat a single linear function. The duration can’t live in a local variable because the start (onCall) and the end (onCallRespond) run at different points in the pipeline — so the start time is stashed on call.attributes under a private AttributeKey<Long> and read back when the response is about to be sent.

src/main/kotlin/com/example/plugins/ResponseTimingPlugin.kt

private val responseTimeStartKey = AttributeKey<Long>("ResponseTimeStart")

val ResponseTimingPlugin = createApplicationPlugin(name = "ResponseTiming") {
    onCall { call ->
        call.attributes.put(responseTimeStartKey, System.currentTimeMillis())
    }

    onCallRespond { call, _ ->
        val startTime = call.attributes.getOrNull(responseTimeStartKey) ?: return@onCallRespond
        val duration = System.currentTimeMillis() - startTime
        call.response.headers.append("X-Response-Time", "${duration}ms")
    }
}

getOrNull(...) returns Long?, and the ?: return@onCallRespond bails out cleanly if the start key was never set — no !!, no crash on an edge case.

RequestIdPlugin.kt — a configurable plugin

Pass createConfiguration = ::RequestIdPluginConfig to createApplicationPlugin and the body gains a pluginConfig of that type. Here the config exposes a header name and an overridable generator (() -> String), so callers can customize the header or swap the UUID generator without touching the plugin. The single onCall hook preserves a client-supplied ID or mints a new one, exposes it to routes via an attribute, and echoes it on the response.

src/main/kotlin/com/example/plugins/RequestIdPlugin.kt

val requestIdAttributeKey = AttributeKey<String>("RequestId")

/**
 * Configuration class for the RequestId plugin.
 * Allows customizing the header name and ID generator.
 */
class RequestIdPluginConfig {
    var headerName: String = "X-Request-Id"
    var generator: () -> String = { UUID.randomUUID().toString() }
}

val RequestIdPlugin = createApplicationPlugin(
    name = "RequestId",
    createConfiguration = ::RequestIdPluginConfig
) {
    val headerName = pluginConfig.headerName
    val generator = pluginConfig.generator

    onCall { call ->
        val requestId = call.request.headers[headerName] ?: generator()
        call.attributes.put(requestIdAttributeKey, requestId)
        call.response.headers.append(headerName, requestId)
    }
}

requestIdAttributeKey is declared at the top level (not private) so routes can read it: in Application.kt the root route does call.attributes.getOrNull(requestIdAttributeKey) to surface the ID in its JSON body. That’s the Ktor analogue of attaching req.requestId in Express or stuffing the ID into the request context.Context in Go.

RateLimitPlugin.kt — stateful, and short-circuiting

A rate limiter needs state that outlives any single request, so the plugin body holds a ConcurrentHashMap<String, RateLimitEntry> keyed by client (default: the remote host). The map is created once when the plugin is installed and closed over by the onCall hook — that closure is the plugin’s instance state. Each entry uses an AtomicInteger count and a @Volatile window-start timestamp for safe concurrent access.

The key move: when the count exceeds the limit, the hook calls call.respond(...) itself. Responding inside onCall short-circuits the pipeline — the route handler never runs. That’s how Ktor expresses “reject this request now,” the equivalent of returning early before next() in Express.

src/main/kotlin/com/example/plugins/RateLimitPlugin.kt

@Serializable
data class RateLimitError(
    val status: Int = 429,
    val error: String = "Too Many Requests",
    val message: String
)

class RateLimitPluginConfig {
    /** Maximum requests per window per IP */
    var maxRequests: Int = 100
    /** Window duration in milliseconds */
    var windowMs: Long = 60_000
    /** Custom key extractor (default: remote host) */
    var keyExtractor: (ApplicationCall) -> String = { call ->
        call.request.local.remoteHost
    }
}

private data class RateLimitEntry(
    val count: AtomicInteger = AtomicInteger(0),
    @Volatile var windowStart: Long = System.currentTimeMillis()
)

val RateLimitPlugin = createApplicationPlugin(
    name = "RateLimit",
    createConfiguration = ::RateLimitPluginConfig
) {
    val maxRequests = pluginConfig.maxRequests
    val windowMs = pluginConfig.windowMs
    val keyExtractor = pluginConfig.keyExtractor
    val clients = ConcurrentHashMap<String, RateLimitEntry>()

    onCall { call ->
        val key = keyExtractor(call)
        val now = System.currentTimeMillis()

        val entry = clients.getOrPut(key) { RateLimitEntry() }

        // Reset window if expired
        if (now - entry.windowStart > windowMs) {
            entry.count.set(0)
            entry.windowStart = now
        }

        val currentCount = entry.count.incrementAndGet()

        // Add rate limit headers
        call.response.headers.append("X-RateLimit-Limit", maxRequests.toString())
        call.response.headers.append("X-RateLimit-Remaining", maxOf(0, maxRequests - currentCount).toString())
        call.response.headers.append("X-RateLimit-Reset", ((entry.windowStart + windowMs) / 1000).toString())

        if (currentCount > maxRequests) {
            val retryAfterSeconds = ((entry.windowStart + windowMs - now) / 1000) + 1
            call.response.headers.append("Retry-After", retryAfterSeconds.toString())
            call.respond(
                HttpStatusCode.TooManyRequests,
                RateLimitError(message = "Rate limit exceeded. Try again in $retryAfterSeconds seconds.")
            )
        }
    }
}

In-memory only

This limiter keeps state in a single process’s ConcurrentHashMap. In production behind multiple instances you’d back it with Redis or a dedicated distributed limiter so the count is shared across nodes.

ApiKeyAuthPlugin.kt — extending Ktor’s Authentication plugin

The first three plugins are standalone. API-key auth instead hooks into Ktor’s built-in Authentication plugin so it composes with authenticate("…") { } route blocks and call.principal<T>(), exactly like the framework’s Bearer/Basic/JWT providers.

The integration point is an extension function on AuthenticationConfig — fun AuthenticationConfig.apiKey(...) — that registers a custom AuthenticationProvider. Inside onAuthenticate you read the header, and on failure call context.challenge(...) to respond with a 401 and complete the challenge; on success call context.principal(...) with your custom Principal.

src/main/kotlin/com/example/plugins/ApiKeyAuthPlugin.kt

@Serializable
data class AuthError(
    val status: Int = 401,
    val error: String = "Unauthorized",
    val message: String
)

/** Represents an authenticated API client */
data class ApiKeyPrincipal(val clientName: String, val apiKey: String) : Principal

class ApiKeyAuthConfig {
    var headerName: String = "X-API-Key"
    internal var validKeys: Map<String, String> = emptyMap()  // key -> clientName

    /** Register valid API keys: map of key to client name */
    fun keys(vararg pairs: Pair<String, String>) {
        validKeys = pairs.toMap()
    }
}

fun AuthenticationConfig.apiKey(
    name: String = "api-key",
    configure: ApiKeyAuthConfig.() -> Unit
) {
    val config = ApiKeyAuthConfig().apply(configure)

    register(object : AuthenticationProvider(object : Config(name) {}) {
        override suspend fun onAuthenticate(context: AuthenticationContext) {
            val call = context.call
            val apiKey = call.request.headers[config.headerName]

            if (apiKey == null) {
                context.challenge("ApiKey", AuthenticationFailedCause.NoCredentials) { challenge, c ->
                    c.respond(
                        HttpStatusCode.Unauthorized,
                        AuthError(message = "Missing ${config.headerName} header")
                    )
                    challenge.complete()
                }
                return
            }

            val clientName = config.validKeys[apiKey]
            if (clientName == null) {
                context.challenge("ApiKey", AuthenticationFailedCause.InvalidCredentials) { challenge, c ->
                    c.respond(
                        HttpStatusCode.Unauthorized,
                        AuthError(message = "Invalid API key")
                    )
                    challenge.complete()
                }
                return
            }

            context.principal(ApiKeyPrincipal(clientName = clientName, apiKey = apiKey))
        }
    })
}

configure: ApiKeyAuthConfig.() -> Unit is a receiver lambda — the same DSL trick that lets you write apiKey("api-key-auth") { headerName = "X-API-Key"; keys(...) } with bare property assignments. The two context.challenge(...) branches distinguish “no credentials” from “bad credentials” so the framework can report the right AuthenticationFailedCause; both respond 401 and call challenge.complete() to stop the pipeline.

Application.kt — installing everything

The module installs the framework plugins (ContentNegotiation, StatusPages) and then the four custom ones. Note the configuration blocks: RequestIdPlugin, RateLimitPlugin, and the apiKey(...) provider all take a trailing install(X) { } lambda that drives their config class. Routes then read what the plugins set — call.attributes.getOrNull(requestIdAttributeKey) for the request ID, and call.principal<ApiKeyPrincipal>() inside an authenticate("…") { } block for the authenticated client.

src/main/kotlin/com/example/plugins/Application.kt

fun main() {
    embeddedServer(Netty, port = 8080) {
        module()
    }.start(wait = true)
}

fun Application.module() {
    // JSON serialization
    install(ContentNegotiation) {
        json(Json {
            prettyPrint = true
            encodeDefaults = true
        })
    }

    // Error handling
    install(StatusPages) {
        exception<Throwable> { call, cause ->
            call.respond(
                HttpStatusCode.InternalServerError,
                mapOf("error" to "Internal Server Error", "message" to (cause.message ?: "Unknown error"))
            )
        }
    }

    install(ResponseTimingPlugin)

    install(RequestIdPlugin) {
        headerName = "X-Request-Id"
    }

    install(RateLimitPlugin) {
        maxRequests = 10           // 10 requests per window (low for demo purposes)
        windowMs = 60_000          // 1 minute window
    }

    // Custom auth: API Key
    install(Authentication) {
        apiKey("api-key-auth") {
            headerName = "X-API-Key"
            keys(
                "demo-key-123" to "demo-client",
                "admin-key-456" to "admin-client"
            )
        }
    }

    routing {
        // Public -- demonstrates ResponseTiming and RequestId
        get("/") {
            val requestId = call.attributes.getOrNull(requestIdAttributeKey) ?: "unknown"
            call.respond(PluginDemo(
                message = "Ktor Custom Plugins Demo",
                requestId = requestId,
                plugins = listOf(
                    "ResponseTiming - adds X-Response-Time header",
                    "RequestId - adds X-Request-Id header",
                    "RateLimit - limits to 10 requests/minute",
                    "ApiKeyAuth - protects /api/protected with X-API-Key header"
                )
            ))
        }

        // Public -- rate limiting demo
        get("/api/public") {
            call.respond(mapOf("message" to "This endpoint is rate-limited to 10 requests/minute"))
        }

        // Protected -- requires API key
        authenticate("api-key-auth") {
            get("/api/protected") {
                val principal = call.principal<ApiKeyPrincipal>()!!
                call.respond(mapOf(
                    "message" to "Hello, ${principal.clientName}! You have access.",
                    "client" to principal.clientName
                ))
            }

            get("/api/protected/admin") {
                val principal = call.principal<ApiKeyPrincipal>()!!
                if (principal.clientName != "admin-client") {
                    call.respond(
                        HttpStatusCode.Forbidden,
                        mapOf("error" to "Forbidden", "message" to "Admin access required")
                    )
                    return@get
                }
                call.respond(mapOf("message" to "Admin area", "client" to principal.clientName))
            }
        }
    }
}

@Serializable
data class PluginDemo(
    val message: String,
    val requestId: String,
    val plugins: List<String>
)

Install order matters: plugins’ onCall hooks fire in the order they were installed, so ResponseTiming starts its timer first and RequestId / RateLimit run before any route handler. /api/protected/admin also shows in-handler authorization — it has a valid key (auth passed) but still returns 403 unless the client is admin-client.

Run it

1. Run the demo server:

   ./gradlew run

2. Hit the root endpoint and inspect the headers — you should see both X-Response-Time and X-Request-Id:

   curl -v http://localhost:8080/

3. Provide your own request ID and watch it get echoed back:

   curl -H "X-Request-Id: my-trace-123" http://localhost:8080/

4. Trip the rate limiter (the demo limit is 10/minute) — the 11th and 12th requests return 429:

   for i in {1..12}; do
     echo "Request $i:"; curl -s -o /dev/null -w "%{http_code}\n" http://localhost:8080/api/public
   done

5. Exercise the API-key auth — missing/invalid keys 401, a valid key 200, and a non-admin key against the admin route 403:

   curl http://localhost:8080/api/protected                                     # 401
   curl -H "X-API-Key: invalid" http://localhost:8080/api/protected             # 401
   curl -H "X-API-Key: demo-key-123" http://localhost:8080/api/protected        # 200
   curl -H "X-API-Key: admin-key-456" http://localhost:8080/api/protected/admin # 200
   curl -H "X-API-Key: demo-key-123" http://localhost:8080/api/protected/admin  # 403

Test it

Each plugin has a focused test under src/test/kotlin/com/example/plugins/ using Ktor’s testApplication host (no real network needed):

./gradlew test

Stretch goals

• Add a CachingPlugin that caches GET responses in memory for a configurable TTL.
• Add a RequestBodyLoggingPlugin that logs incoming request bodies for debugging (use the onCallReceive hook).
• Add an IPWhitelistPlugin that only allows requests from specified IP addresses.
• Modify RateLimitPlugin to use different limits for different route prefixes.
• Add a CompressionPlugin wrapper that enables gzip for responses larger than 1KB.