Usage

PluginHandler

Implement PluginHandler to define your plugin's behavior:

from kbs_plugin_sdk import PluginHandler, PluginRequest, PluginResponse
from kbs_plugin_sdk.plugin.plugin_pb2 import ValidateAuthRequest, NeedsEncryptionRequest

class MyPlugin(PluginHandler):
    async def handle(self, request: PluginRequest) -> PluginResponse:
        return PluginResponse(body=b"ok", status_code=200)

    async def validate_auth(self, request: ValidateAuthRequest) -> bool:
        return request.method == "POST"

    async def needs_encryption(self, request: NeedsEncryptionRequest) -> bool:
        return True

Method	Default	Description
`handle(request)`	required	Process the request and return a response
`validate_auth(request)`	`False`	Return `True` to require KBS admin authentication
`needs_encryption(request)`	`True`	Return `True` to JWE-encrypt the response with the TEE public key

PluginRequest fields

Field	Type	Description
`method`	`str`	HTTP method (`GET`, `POST`, etc.)
`path`	sequence of `str`	URL path segments
`query`	mapping of `str → str`	URL query parameters
`body`	`bytes`	Request body

PluginResponse fields

Field	Type	Description
`body`	`bytes`	Response body
`status_code`	`int`	HTTP status code
`content_type`	`str`	Content-Type header (optional, default `""`)

Authentication note

validate_auth controls whether KBS admin credentials are required. GET requests on the external plugin path always require a valid client attestation token from the caller, regardless of what validate_auth returns. In development with InsecureAllowAll admin mode, POST requests work without credentials; GET requests still require real TEE attestation.

PluginServer

PluginServer uses a fluent builder API:

from kbs_plugin_sdk import PluginServer, TlsConfig

server = (
    PluginServer(my_handler)
    .with_address("0.0.0.0:50051")
    .with_tls(TlsConfig.server_tls("server.crt", "server.key"))
)
await server.serve()

Method	Description
`with_address(address)`	Bind address, default `127.0.0.1:50051`
`with_tls(tls_config)`	Enable server-side TLS
`without_health()`	Disable the gRPC health service
`serve()`	Start the server (async, blocks until SIGTERM/SIGINT)

The server automatically registers gRPC reflection and health checking. It handles SIGTERM and SIGINT with a 5-second grace period.

TLS

KBS supports server-side TLS and insecure (plaintext) connections. Mutual TLS is not supported by KBS.

from kbs_plugin_sdk import TlsConfig

tls = TlsConfig.server_tls("server.crt", "server.key")
server = PluginServer(handler).with_address("0.0.0.0:50051").with_tls(tls)
await server.serve()

Starting without TLS emits a warning at startup and should only be used for local development.

TlsConfig

Method	Description
`TlsConfig.server_tls(cert, key)`	Create config from PEM cert and key file paths