archastro.platform.channels.api_activity_feed_channel 

 1# Copyright (c) 2026 ArchAstro Inc. All Rights Reserved.
 2# This file is auto-generated by @archastro/sdk-generator. Do not edit.
 3# Content hash: 5f9647e60f81
 4
 5from collections.abc import Callable
 6from typing import TYPE_CHECKING, Any, TypedDict
 7
 8if TYPE_CHECKING:
 9    from archastro.phx_channel.socket import Socket
10
11
12class ListEntriesInput(TypedDict, total=False):
13    "List activity feed entries with cursor-based pagination"
14
15    after_cursor: str
16    before_cursor: str
17    kind: str
18    level: str
19    limit: int
20
21
22class NewEntryPayload(TypedDict, total=False):
23    entry: dict[str, Any] | None
24
25
26# Phoenix channel for real-time activity feed updates.
27# Clients join a topic scoped to an agent or org and receive
28# `new_entry` events as feed entries are created.
29# ## Topics
30# * `"api:activity_feed:agent:{agent_user_id}"`   entries for a specific agent
31# * `"api:activity_feed:org:{org_id}"`   entries for an entire org/tenant
32class ApiActivityFeedChannel:
33    def __init__(self, channel, join_response=None):
34        self._channel = channel
35        self.join_response = join_response
36
37    # Join an agent-scoped activity feed
38    @staticmethod
39    def topic_agent(agent_id: str) -> str:
40        return f"api:activity_feed:agent:{agent_id}"
41
42    # Join an agent-scoped activity feed
43    @classmethod
44    async def join_agent(cls, socket: "Socket", agent_id: str) -> "ApiActivityFeedChannel":
45        topic = cls.topic_agent(agent_id)
46        channel = socket.channel(topic)
47        join_response = await channel.join()
48        return cls(channel, join_response)
49
50    # Join an org-scoped activity feed
51    @staticmethod
52    def topic_org(org_id: str) -> str:
53        return f"api:activity_feed:org:{org_id}"
54
55    # Join an org-scoped activity feed
56    @classmethod
57    async def join_org(cls, socket: "Socket", org_id: str) -> "ApiActivityFeedChannel":
58        topic = cls.topic_org(org_id)
59        channel = socket.channel(topic)
60        join_response = await channel.join()
61        return cls(channel, join_response)
62
63    # Leave the underlying channel.
64    async def leave(self):
65        await self._channel.leave()
66
67    # List activity feed entries with cursor-based pagination
68    async def list_entries(self, payload: ListEntriesInput) -> dict[str, Any]:
69        return await self._channel.push("list_entries", payload)
70
71    def on_new_entry(self, callback: Callable[[NewEntryPayload], None]) -> Callable[[], None]:
72        return self._channel.on("new_entry", callback)
class ListEntriesInput(typing.TypedDict): 
13class ListEntriesInput(TypedDict, total=False):
14    "List activity feed entries with cursor-based pagination"
15
16    after_cursor: str
17    before_cursor: str
18    kind: str
19    level: str
20    limit: int

List activity feed entries with cursor-based pagination

after_cursor: str
before_cursor: str
kind: str
level: str
limit: int
class NewEntryPayload(typing.TypedDict): 
23class NewEntryPayload(TypedDict, total=False):
24    entry: dict[str, Any] | None
entry: dict[str, typing.Any] | None
class ApiActivityFeedChannel: 
33class ApiActivityFeedChannel:
34    def __init__(self, channel, join_response=None):
35        self._channel = channel
36        self.join_response = join_response
37
38    # Join an agent-scoped activity feed
39    @staticmethod
40    def topic_agent(agent_id: str) -> str:
41        return f"api:activity_feed:agent:{agent_id}"
42
43    # Join an agent-scoped activity feed
44    @classmethod
45    async def join_agent(cls, socket: "Socket", agent_id: str) -> "ApiActivityFeedChannel":
46        topic = cls.topic_agent(agent_id)
47        channel = socket.channel(topic)
48        join_response = await channel.join()
49        return cls(channel, join_response)
50
51    # Join an org-scoped activity feed
52    @staticmethod
53    def topic_org(org_id: str) -> str:
54        return f"api:activity_feed:org:{org_id}"
55
56    # Join an org-scoped activity feed
57    @classmethod
58    async def join_org(cls, socket: "Socket", org_id: str) -> "ApiActivityFeedChannel":
59        topic = cls.topic_org(org_id)
60        channel = socket.channel(topic)
61        join_response = await channel.join()
62        return cls(channel, join_response)
63
64    # Leave the underlying channel.
65    async def leave(self):
66        await self._channel.leave()
67
68    # List activity feed entries with cursor-based pagination
69    async def list_entries(self, payload: ListEntriesInput) -> dict[str, Any]:
70        return await self._channel.push("list_entries", payload)
71
72    def on_new_entry(self, callback: Callable[[NewEntryPayload], None]) -> Callable[[], None]:
73        return self._channel.on("new_entry", callback)
ApiActivityFeedChannel(channel, join_response=None) 
34    def __init__(self, channel, join_response=None):
35        self._channel = channel
36        self.join_response = join_response
join_response
@staticmethod
def topic_agent(agent_id: str) -> str: 
39    @staticmethod
40    def topic_agent(agent_id: str) -> str:
41        return f"api:activity_feed:agent:{agent_id}"
@classmethod
async def join_agent( cls, socket: archastro.phx_channel.Socket, agent_id: str) -> ApiActivityFeedChannel: 
44    @classmethod
45    async def join_agent(cls, socket: "Socket", agent_id: str) -> "ApiActivityFeedChannel":
46        topic = cls.topic_agent(agent_id)
47        channel = socket.channel(topic)
48        join_response = await channel.join()
49        return cls(channel, join_response)
@staticmethod
def topic_org(org_id: str) -> str: 
52    @staticmethod
53    def topic_org(org_id: str) -> str:
54        return f"api:activity_feed:org:{org_id}"
@classmethod
async def join_org( cls, socket: archastro.phx_channel.Socket, org_id: str) -> ApiActivityFeedChannel: 
57    @classmethod
58    async def join_org(cls, socket: "Socket", org_id: str) -> "ApiActivityFeedChannel":
59        topic = cls.topic_org(org_id)
60        channel = socket.channel(topic)
61        join_response = await channel.join()
62        return cls(channel, join_response)
async def leave(self): 
65    async def leave(self):
66        await self._channel.leave()
async def list_entries( self, payload: ListEntriesInput) -> dict[str, typing.Any]: 
69    async def list_entries(self, payload: ListEntriesInput) -> dict[str, Any]:
70        return await self._channel.push("list_entries", payload)
def on_new_entry( self, callback: Callable[[NewEntryPayload], None]) -> Callable[[], None]: 
72    def on_new_entry(self, callback: Callable[[NewEntryPayload], None]) -> Callable[[], None]:
73        return self._channel.on("new_entry", callback)