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.8 kB · 91 lines
Python
at main
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192from __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 intakeUrl: str | None signatureHeader: str signaturePrefix: str revision: int
host: Host
class Webhooks: async def list(self) -> builtins.list[Hook]: """Hooks targeting this session; a sender posts to `intakeUrl`. 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}