Managed echo example

This is a complete minimal unary child entrypoint. Copy it into your own application module. It needs the SDK wheel and an installed manifest declaring exactly the ECHO descriptor. It does not start a host or create a signed package on its own.

import argparse
import asyncio
from pydantic import JsonValue
from skulk_capability_sdk.contracts import (
    Descriptor, Health, Hello, Invoke, Result, Shutdown, Startup,
    manifest_digest,
)
from skulk_capability_sdk.storage import read_startup
from skulk_capability_sdk.wire import read_message, write_message

TEXT_SCHEMA: dict[str, JsonValue] = {
    "type": "object",
    "properties": {"text": {"type": "string", "maxLength": 1024}},
    "required": ["text"],
    "additionalProperties": False,
}
ECHO = Descriptor(
    id="echo", version="1.0.0", title="Echo",
    description="Return up to 1024 characters of input text unchanged.",
    input_schema=TEXT_SCHEMA, output_schema=TEXT_SCHEMA,
)


def answer(call: Invoke) -> Result:
    """Validate direct input and return one correlated outcome."""
    matches = (
        call.capability_id == ECHO.id
        and call.version == ECHO.version
        and call.descriptor_revision == ECHO.revision
    )
    text = call.payload.get("text")
    if not matches or set(call.payload) != {"text"} or not isinstance(text, str) or len(text) > 1024:
        return Result(call_id=call.call_id, payload=None, error="invalid_payload")
    return Result(call_id=call.call_id, payload={"text": text}, error=None)


async def run(startup: Startup) -> None:
    """Serve the inherited host connection until shutdown or connection loss."""
    if startup.manifest.descriptors != (ECHO,):
        raise ValueError("This example requires the exact echo manifest")
    reader, writer = await asyncio.open_unix_connection(startup.socket)
    try:
        await write_message(writer, Hello(
            host_id=startup.host_id,
            transport_node_id=startup.transport_id,
            capability_node_id=startup.node_id,
            bundle_id=startup.manifest.bundle_id,
            bundle_version=startup.manifest.bundle_version,
            token=startup.token,
            manifest_sha256=manifest_digest(startup.manifest),
        ))
        while True:
            message = await read_message(reader)
            if isinstance(message, Shutdown):
                return
            if isinstance(message, Health):
                await write_message(writer, Health(nonce=message.nonce, ready=True))
            elif isinstance(message, Invoke):
                async with asyncio.timeout(message.remaining_seconds):
                    await write_message(writer, answer(message))
    finally:
        writer.close()
        await writer.wait_closed()


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--config-fd", type=int, required=True)
    args = parser.parse_args()
    asyncio.run(run(read_startup(args.config_fd)))

The host supplies --config-fd; do not invent that record or put its token in a shell command. The echo operation is immediate and does not block health behind a long-running job. Unsupported message modes are not implemented; the signed manifest advertises unary only.

Check the pure operation first

request = Invoke(
    call_id="example-call", capability_id=ECHO.id, version=ECHO.version,
    descriptor_revision=ECHO.revision, remaining_seconds=5.0,
    payload={"text": "Hello from the fabric."},
)
assert answer(request).payload == {"text": "Hello from the fabric."}
assert answer(request.model_copy(update={"payload": {"text": 42}})).error == "invalid_payload"

Then qualify a real installed child: startup rejection on the wrong descriptor, a successful call through Skulk™, failed negotiation, health, graceful shutdown and crash recovery. Manifest and qualification explain that next boundary.

Check the pure operation first​

Check the pure operation first