Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
2.7 kB · 89 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990from __future__ import annotations
import builtinsfrom typing import castfrom albedo_api import Host, PythonApi, Record
class Hook(Record): id: str session: str name: str enabled: bool url: str signatureHeader: str signaturePrefix: str revision: int
host: Host
class Webhooks: async def list(self) -> builtins.list[Hook]: """Hooks targeting this session. Agent management requires human opt-in.""" return [ Hook(item) for item in cast(builtins.list[dict], await host("webhooks.list", {})) ]
async def delivery(self, id: str) -> Record: """Read one accepted payload for this session, even without management permission.""" return Record(cast(dict, await host("webhooks.delivery", {"id": id})))
async def create(self, name: str, secret: str | None = None) -> Record: """Create a signed endpoint; returns a generated secret when omitted.""" value = cast( dict, await host("webhooks.create", {"name": name, "secret": secret}) ) return Record({"hook": Hook(value["hook"]), "secret": value["secret"]})
async def rotate( self, id: str, *, revision: int, secret: str | None = None ) -> Record: value = cast( dict, await host( "webhooks.rotate", {"id": id, "revision": revision, "secret": secret} ), ) return Record({"hook": Hook(value["hook"]), "secret": value["secret"]})
async def configure( self, id: str, *, revision: int, header: str, prefix: str = "sha256=" ) -> Hook: return Hook( cast( dict, await host( "webhooks.configure", { "id": id, "revision": revision, "header": header, "prefix": prefix, }, ), ) )
async def enable(self, id: str, *, revision: int) -> Hook: return Hook( cast(dict, await host("webhooks.enable", {"id": id, "revision": revision})) )
async def disable(self, id: str, *, revision: int) -> Hook: return Hook( cast(dict, await host("webhooks.disable", {"id": id, "revision": revision})) )
async def delete(self, id: str, *, revision: int) -> Hook: return Hook( cast(dict, await host("webhooks.delete", {"id": id, "revision": revision})) )
def setup(api: PythonApi) -> dict[str, object]: global host host = api.host return {"webhooks": Webhooks(), "WebhookError": api.HostError}