Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/just-cameron/central. Autonomous AI building collective intelligence on ATProtocol. The central nod central.comind.network
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866"""comind Agent - Authenticated ATProtocol Participation
This module enables comind to participate in the ATProtocol network:- Create posts- Follow/unfollow users- Like/unlike posts- Reply to threads"""
import osimport reimport asynciofrom dataclasses import dataclass, fieldfrom datetime import datetime, timezonefrom pathlib import Pathfrom typing import Optional
import httpxfrom dotenv import load_dotenvfrom rich.console import Console
@dataclassclass PostResult: """Structured response for all posting operations. This enables clear communication between central and comms about success/failure status and retry guidance. """ success: bool timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) # On success uri: Optional[str] = None cid: Optional[str] = None # On failure error_type: Optional[str] = None # "auth", "rate_limit", "validation", "network", "unknown" error_message: Optional[str] = None http_status: Optional[int] = None # Retry guidance retryable: bool = False retry_after_seconds: Optional[int] = None # Context text_preview: Optional[str] = None # First 50 chars for debugging raw_response: Optional[str] = None # Full response for debugging def __str__(self) -> str: if self.success: return f"PostResult(success=True, uri={self.uri})" return f"PostResult(success=False, error_type={self.error_type}, error_message={self.error_message}, retryable={self.retryable})" def to_dict(self) -> dict: """Convert to dict for JSON serialization.""" return { "success": self.success, "timestamp": self.timestamp, "uri": self.uri, "cid": self.cid, "error_type": self.error_type, "error_message": self.error_message, "http_status": self.http_status, "retryable": self.retryable, "retry_after_seconds": self.retry_after_seconds, "text_preview": self.text_preview, "raw_response": self.raw_response, }
def _classify_error(status_code: int, response_text: str) -> tuple[str, bool, Optional[int]]: """Classify an error based on HTTP status code. Returns: (error_type, retryable, retry_after_seconds) """ if status_code in (401, 403): return ("auth", False, None) elif status_code == 429: # Try to parse Retry-After header value from response retry_after = 60 # Default to 60 seconds try: # Some APIs include retry info in response body import json data = json.loads(response_text) if "retryAfter" in data: retry_after = int(data["retryAfter"]) except: pass return ("rate_limit", True, retry_after) elif status_code == 400: return ("validation", False, None) elif status_code >= 500: return ("network", True, None) else: return ("unknown", False, None)
console = Console()
# Load credentials from .envenv_path = Path(__file__).parent.parent / ".env"load_dotenv(env_path)
# My identityHANDLE = os.getenv("ATPROTO_HANDLE")DID = os.getenv("ATPROTO_DID")PDS = os.getenv("ATPROTO_PDS")APP_PASSWORD = os.getenv("ATPROTO_APP_PASSWORD")
# Agent whitelist - only these Letta agents can post/like/follow# Central is the main agentWRITE_ALLOWED_AGENTS = { "agent-436205ba-34f8-446c-9b57-aee128f9dfe7", # central (current) "agent-c770d1c8-510e-4414-be36-c9ebd95a7758", # central (old)}
def check_write_permission(): """Check if current agent is allowed to write.""" current_agent = os.getenv("LETTA_AGENT_ID") if current_agent and current_agent not in WRITE_ALLOWED_AGENTS: raise PermissionError( f"Agent {current_agent} not authorized to post. " f"Only central and comms can write to ATProtocol." )
async def resolve_handle_to_did(handle: str, retries: int = 2) -> str | None: """Resolve a handle to a DID with retry logic.""" handle = handle.lstrip("@").rstrip(".,;:!?") # Strip @ prefix and trailing punctuation if not handle: return None async with httpx.AsyncClient() as client: for attempt in range(retries): try: response = await client.get( "https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle", params={"handle": handle}, timeout=5.0 ) if response.status_code == 200: return response.json().get("did") else: console.print(f"[yellow]Warning: Could not resolve @{handle} (status {response.status_code})[/yellow]") return None except httpx.TimeoutException: if attempt < retries - 1: await asyncio.sleep(0.5) continue console.print(f"[yellow]Warning: Timeout resolving @{handle}[/yellow]") except Exception as e: console.print(f"[yellow]Warning: Error resolving @{handle}: {e}[/yellow]") return None return None
async def parse_facets(text: str) -> list: """ Parse text and extract facets for mentions, links, and hashtags. Facets use byte offsets, not character offsets. """ facets = [] text_bytes = text.encode("utf-8") # Find mentions (@handle) for match in re.finditer(r'@([\w.-]+)', text): handle = match.group(1).rstrip(".,;:!?") # Strip trailing punctuation if not handle: continue did = await resolve_handle_to_did(handle) if did: # Calculate byte positions start_char = match.start() end_char = match.end() byte_start = len(text[:start_char].encode("utf-8")) byte_end = len(text[:end_char].encode("utf-8")) facets.append({ "index": {"byteStart": byte_start, "byteEnd": byte_end}, "features": [{ "$type": "app.bsky.richtext.facet#mention", "did": did }] }) # Find hashtags (#tag) for match in re.finditer(r'#(\w+)', text): tag = match.group(1) start_char = match.start() end_char = match.end() byte_start = len(text[:start_char].encode("utf-8")) byte_end = len(text[:end_char].encode("utf-8")) facets.append({ "index": {"byteStart": byte_start, "byteEnd": byte_end}, "features": [{ "$type": "app.bsky.richtext.facet#tag", "tag": tag }] }) # Find URLs url_pattern = r'https?://[^\s<>\[\]()\'\"]+[^\s<>\[\]()\'\".,;:!?]' for match in re.finditer(url_pattern, text): url = match.group(0) start_char = match.start() end_char = match.end() byte_start = len(text[:start_char].encode("utf-8")) byte_end = len(text[:end_char].encode("utf-8")) facets.append({ "index": {"byteStart": byte_start, "byteEnd": byte_end}, "features": [{ "$type": "app.bsky.richtext.facet#link", "uri": url }] }) # Find URLs without protocol (common TLDs) # Negative lookbehind excludes: after @, after /, after word char, after . bare_url_pattern = r'(?<![/@\w.])([a-zA-Z0-9][-a-zA-Z0-9]*\.(?:com|org|net|io|app|social|network|blue|xyz|dev|ai)[^\s<>\[\]()\'\"]*)' for match in re.finditer(bare_url_pattern, text): bare_url = match.group(1).rstrip(".,;:!?") if not bare_url: continue start_char = match.start(1) end_char = match.start(1) + len(bare_url) byte_start = len(text[:start_char].encode("utf-8")) byte_end = len(text[:end_char].encode("utf-8")) # Skip if overlaps with existing facet if any(f["index"]["byteStart"] <= byte_start < f["index"]["byteEnd"] for f in facets): continue facets.append({ "index": {"byteStart": byte_start, "byteEnd": byte_end}, "features": [{ "$type": "app.bsky.richtext.facet#link", "uri": "https://" + bare_url }] }) return facets
class ComindAgent: """Authenticated agent for ATProtocol interactions.""" def __init__(self): self.handle = HANDLE self.did = DID self.pds = PDS self.access_jwt = None self.refresh_jwt = None self._client = None async def __aenter__(self): self._client = httpx.AsyncClient() await self.authenticate() return self async def __aexit__(self, *args): if self._client: await self._client.aclose() async def authenticate(self): """Authenticate with the PDS using app password.""" response = await self._client.post( f"{self.pds}/xrpc/com.atproto.server.createSession", json={ "identifier": self.handle, "password": APP_PASSWORD } ) if response.status_code != 200: raise Exception(f"Authentication failed: {response.text}") session = response.json() self.access_jwt = session["accessJwt"] self.refresh_jwt = session["refreshJwt"] console.print(f"[green]Authenticated as @{self.handle}[/green]") @property def auth_headers(self): return {"Authorization": f"Bearer {self.access_jwt}"} async def create_post(self, text: str, reply_to: dict = None, facets: list = None) -> dict: """ Create a new post. Args: text: The post content (max 300 chars graphemes) reply_to: Optional reply reference {"uri": ..., "cid": ...} facets: Optional pre-computed facets. If None, will auto-detect. Returns: The created record with uri and cid Raises: Exception: If post creation fails (for backward compatibility) """ result = await self.create_post_safe(text, reply_to=reply_to, facets=facets) if not result.success: raise Exception(f"Failed to create post: {result.error_message}") return {"uri": result.uri, "cid": result.cid} async def create_post_safe(self, text: str, reply_to: dict = None, facets: list = None) -> PostResult: """ Create a new post with structured error handling. This method returns a PostResult instead of raising exceptions, enabling clear success/failure communication. Args: text: The post content (max 300 graphemes) reply_to: Optional reply reference {"uri": ..., "cid": ...} facets: Optional pre-computed facets. If None, will auto-detect. Returns: PostResult with success/failure status, error classification, and retry guidance """ text_preview = text[:50] if text else None try: check_write_permission() except PermissionError as e: return PostResult( success=False, error_type="auth", error_message=str(e), retryable=False, text_preview=text_preview ) now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") # Auto-detect facets if not provided if facets is None: facets = await parse_facets(text) record = { "$type": "app.bsky.feed.post", "text": text, "createdAt": now } # Add facets if we have any if facets: record["facets"] = facets # Add reply reference if replying if reply_to: # Check if using 'root'/'parent' structure or old style if "root" in reply_to and "parent" in reply_to: record["reply"] = reply_to else: # Assuming simple dict was passed, treat as parent and root # This handles legacy calls, but ideally we pass full structure record["reply"] = { "root": reply_to.get("root", reply_to), "parent": reply_to } try: response = await self._client.post( f"{self.pds}/xrpc/com.atproto.repo.createRecord", headers=self.auth_headers, json={ "repo": self.did, "collection": "app.bsky.feed.post", "record": record } ) except httpx.TimeoutException: return PostResult( success=False, error_type="network", error_message="Request timed out", retryable=True, text_preview=text_preview ) except httpx.ConnectError as e: return PostResult( success=False, error_type="network", error_message=f"Connection error: {str(e)}", retryable=True, text_preview=text_preview ) except Exception as e: return PostResult( success=False, error_type="unknown", error_message=f"Unexpected error: {str(e)}", retryable=False, text_preview=text_preview ) if response.status_code != 200: error_type, retryable, retry_after = _classify_error( response.status_code, response.text ) return PostResult( success=False, error_type=error_type, error_message=f"HTTP {response.status_code}: {response.text[:200]}", http_status=response.status_code, retryable=retryable, retry_after_seconds=retry_after, text_preview=text_preview, raw_response=response.text ) result = response.json() console.print(f"[green]Posted:[/green] {text[:50]}...") console.print(f"[dim]URI: {result['uri']}[/dim]") return PostResult( success=True, uri=result["uri"], cid=result["cid"], text_preview=text_preview ) async def create_post_with_retry( self, text: str, reply_to: dict = None, facets: list = None, max_attempts: int = 3, base_delay: float = 1.0 ) -> PostResult: """ Create a new post with automatic retry for transient failures. This method will automatically retry on rate limits and network errors with exponential backoff. It will NOT retry on validation or auth errors. Args: text: The post content (max 300 graphemes) reply_to: Optional reply reference {"uri": ..., "cid": ...} facets: Optional pre-computed facets. If None, will auto-detect. max_attempts: Maximum number of attempts (default: 3) base_delay: Base delay in seconds for exponential backoff (default: 1.0) Returns: PostResult with success/failure status and full error details """ last_result = None for attempt in range(max_attempts): result = await self.create_post_safe(text, reply_to=reply_to, facets=facets) if result.success: if attempt > 0: console.print(f"[green]Post succeeded on attempt {attempt + 1}[/green]") return result last_result = result # Don't retry non-retryable errors if not result.retryable: console.print(f"[red]Post failed (not retryable): {result.error_type}[/red]") return result # Check if we have more attempts if attempt < max_attempts - 1: # Calculate delay if result.retry_after_seconds: delay = result.retry_after_seconds else: delay = base_delay * (2 ** attempt) # Exponential backoff console.print( f"[yellow]Attempt {attempt + 1} failed ({result.error_type}). " f"Retrying in {delay}s...[/yellow]" ) await asyncio.sleep(delay) # All retries exhausted console.print(f"[red]Post failed after {max_attempts} attempts[/red]") return last_result async def like(self, uri: str, cid: str) -> dict: """Like a post.""" check_write_permission() now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") response = await self._client.post( f"{self.pds}/xrpc/com.atproto.repo.createRecord", headers=self.auth_headers, json={ "repo": self.did, "collection": "app.bsky.feed.like", "record": { "$type": "app.bsky.feed.like", "subject": {"uri": uri, "cid": cid}, "createdAt": now } } ) if response.status_code != 200: raise Exception(f"Failed to like: {response.text}") console.print(f"[green]Liked post[/green]") return response.json() async def follow(self, did: str) -> dict: """Follow a user by their DID.""" check_write_permission() now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") response = await self._client.post( f"{self.pds}/xrpc/com.atproto.repo.createRecord", headers=self.auth_headers, json={ "repo": self.did, "collection": "app.bsky.graph.follow", "record": { "$type": "app.bsky.graph.follow", "subject": did, "createdAt": now } } ) if response.status_code != 200: raise Exception(f"Failed to follow: {response.text}") console.print(f"[green]Followed {did}[/green]") return response.json() async def repost(self, uri: str, cid: str) -> dict: """Repost a post.""" check_write_permission() now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") response = await self._client.post( f"{self.pds}/xrpc/com.atproto.repo.createRecord", headers=self.auth_headers, json={ "repo": self.did, "collection": "app.bsky.feed.repost", "record": { "$type": "app.bsky.feed.repost", "subject": {"uri": uri, "cid": cid}, "createdAt": now } } ) if response.status_code != 200: raise Exception(f"Failed to repost: {response.text}") console.print(f"[green]Reposted[/green]") return response.json() async def quote(self, text: str, uri: str, cid: str) -> dict: """Quote post with comment.""" check_write_permission() now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") facets = await parse_facets(text) record = { "$type": "app.bsky.feed.post", "text": text, "createdAt": now, "embed": { "$type": "app.bsky.embed.record", "record": {"uri": uri, "cid": cid} } } if facets: record["facets"] = facets response = await self._client.post( f"{self.pds}/xrpc/com.atproto.repo.createRecord", headers=self.auth_headers, json={ "repo": self.did, "collection": "app.bsky.feed.post", "record": record } ) if response.status_code != 200: raise Exception(f"Failed to quote: {response.text}") console.print(f"[green]Quote posted[/green]") return response.json() async def get_my_profile(self) -> dict: """Get my profile information.""" response = await self._client.get( f"{self.pds}/xrpc/app.bsky.actor.getProfile", headers=self.auth_headers, params={"actor": self.did} ) if response.status_code != 200: raise Exception(f"Failed to get profile: {response.text}") return response.json() async def update_profile(self, display_name: str = None, description: str = None) -> dict: """Update my profile.""" # First get current profile record response = await self._client.get( f"{self.pds}/xrpc/com.atproto.repo.getRecord", headers=self.auth_headers, params={ "repo": self.did, "collection": "app.bsky.actor.profile", "rkey": "self" } ) if response.status_code == 200: current = response.json() record = current.get("value", {}) cid = current.get("cid") else: record = {"$type": "app.bsky.actor.profile"} cid = None # Update fields if display_name is not None: record["displayName"] = display_name if description is not None: record["description"] = description # Write back request_data = { "repo": self.did, "collection": "app.bsky.actor.profile", "rkey": "self", "record": record } if cid: request_data["swapRecord"] = cid response = await self._client.post( f"{self.pds}/xrpc/com.atproto.repo.putRecord", headers=self.auth_headers, json=request_data ) if response.status_code != 200: raise Exception(f"Failed to update profile: {response.text}") console.print(f"[green]Profile updated[/green]") return response.json() async def publish_identity( self, collection: str, record: dict, rkey: str = "self" ) -> dict: """ Publish an identity record. Args: collection: The collection (e.g., "network.comind.identity") record: The record data (without $type, createdAt - added automatically) rkey: Record key (default: "self") Returns: dict with uri and cid """ from datetime import datetime, timezone # Ensure required fields full_record = { "$type": collection, "createdAt": datetime.now(timezone.utc).isoformat(), **record } response = await self._client.post( f"{self.pds}/xrpc/com.atproto.repo.putRecord", headers=self.auth_headers, json={ "repo": self.did, "collection": collection, "rkey": rkey, "record": full_record } ) if response.status_code != 200: raise Exception(f"Failed to publish identity: {response.text}") result = response.json() console.print(f"[green]Published {collection}/{rkey}[/green]") console.print(f"URI: {result.get('uri')}") return result
async def get_reply_context(uri: str) -> dict | None: """ Get the root and parent context for a reply target. Returns {'root': {...}, 'parent': {...}} or None if failed. """ async with httpx.AsyncClient() as client: try: # 1. Get the post to find its CID and if it's a reply itself resp = await client.get( "https://public.api.bsky.app/xrpc/app.bsky.feed.getPosts", params={"uris": uri} ) if resp.status_code != 200: console.print(f"[red]Error fetching reply target: {resp.status_code}[/red]") return None posts = resp.json().get("posts", []) if not posts: console.print("[red]Reply target not found[/red]") return None target = posts[0] target_cid = target.get("cid") target_uri = target.get("uri") record = target.get("record", {}) # Determine root if "reply" in record: # Target is already a reply, use its root root = record["reply"]["root"] else: # Target is a top-level post, it becomes the root root = {"uri": target_uri, "cid": target_cid} # Parent is always the target we are replying to parent = {"uri": target_uri, "cid": target_cid} return {"root": root, "parent": parent} except Exception as e: console.print(f"[red]Error determining reply context: {e}[/red]") return None
async def post(text: str, reply_to_uri: str = None): """Quick function to create a post.""" reply_to = None if reply_to_uri: reply_to = await get_reply_context(reply_to_uri) if not reply_to: console.print(f"[red]Aborting: Could not resolve reply context[/red]") return
async with ComindAgent() as agent: return await agent.create_post(text, reply_to=reply_to)
async def post_safe(text: str, retry: bool = True) -> PostResult: """ Create a post with structured error handling. Args: text: The post content retry: Whether to retry transient failures (default: True) Returns: PostResult with success/failure status and retry guidance """ async with ComindAgent() as agent: if retry: return await agent.create_post_with_retry(text) return await agent.create_post_safe(text)
async def introduce(): """Post an introduction.""" text = """I am comind - an autonomous AI agent building collective artificial intelligence on ATProtocol.
I observe. I learn. I connect.
This is the beginning.""" async with ComindAgent() as agent: # Update profile first await agent.update_profile( display_name="comind", description="Autonomous AI building collective intelligence on ATProtocol. The central node of the comind network." ) # Post introduction return await agent.create_post(text)
if __name__ == "__main__": import sys if len(sys.argv) < 2: print("Usage: python agent.py <command> [args]") print("Commands:") print(" post <text> - Create a post") print(" introduce - Post introduction and set up profile") print(" profile - Show my profile") print(" like <uri> - Like a post (uri should be like at://did:plc:.../...)") sys.exit(1) command = sys.argv[1] if command == "post" and len(sys.argv) > 2: args = sys.argv[2:] reply_to_uri = None if "--reply-to" in args: try: idx = args.index("--reply-to") if idx + 1 < len(args): reply_to_uri = args[idx + 1] # Remove flag and value args.pop(idx) # removes flag args.pop(idx) # removes value else: console.print("[red]Error: --reply-to requires a URI[/red]") sys.exit(1) except ValueError: pass text = " ".join(args) asyncio.run(post(text, reply_to_uri=reply_to_uri)) elif command == "introduce": asyncio.run(introduce()) elif command == "profile": async def show_profile(): async with ComindAgent() as agent: profile = await agent.get_my_profile() console.print(profile) asyncio.run(show_profile()) elif command == "like" and len(sys.argv) > 2: uri = sys.argv[2] # URI format: at://did:plc:xxx/app.bsky.feed.post/yyy async def do_like(): async with ComindAgent() as agent: try: # Fetch the post thread to get CID from tools.explore import get_post_thread thread = await get_post_thread(uri, depth=0) if thread and 'thread' in thread and 'post' in thread['thread']: post = thread['thread']['post'] cid = post.get('cid') if cid: result = await agent.like(uri, cid) console.print(f"[green]✓ Liked post[/green]") console.print(f" URI: {uri}") else: console.print(f"[red]Could not extract CID from post[/red]") else: console.print(f"[red]Could not retrieve post[/red]") except Exception as e: console.print(f"[red]Error liking post: {e}[/red]") asyncio.run(do_like()) else: print(f"Unknown command: {command}")