Something went wrong. Try again.
Something went wrong. Try again.
Python
at toolchange
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106"""Reply to Bluesky posts tool."""
from typing import Optionalfrom pydantic import BaseModel, Field
class ReplyArgs(BaseModel): """Arguments for replying to a Bluesky post.""" text: str = Field(..., description="The reply text content (max 300 characters)") reply_to_uri: str = Field(..., description="The AT URI of the post being replied to (required)") reply_to_cid: str = Field(..., description="The CID of the post being replied to (required)")
def reply_to_bluesky_post(text: str, reply_to_uri: str, reply_to_cid: str) -> str: """ Reply to an existing Bluesky post. This tool creates a threaded reply to a specific post. IMPORTANT: This tool is ONLY for replying to existing posts. To create a new standalone post, use create_new_bluesky_post instead. Args: text: The reply text content (max 300 characters) reply_to_uri: The AT URI of the post being replied to (required) reply_to_cid: The CID of the post being replied to (required) Returns: Success message with post URI or error message Raises: Exception: If the reply fails """ if len(text) > 300: raise ValueError(f"Reply text too long: {len(text)} characters (max 300)") if not reply_to_uri: raise ValueError("reply_to_uri is required for replies") if not reply_to_cid: raise ValueError("reply_to_cid is required for replies") import os import requests from datetime import datetime, timezone try: # Get credentials from environment username = os.getenv("BSKY_USERNAME") password = os.getenv("BSKY_PASSWORD") pds_host = os.getenv("PDS_URI", "https://bsky.social") if not username or not password: raise Exception("BSKY_USERNAME and BSKY_PASSWORD environment variables must be set") # Create session session_url = f"{pds_host}/xrpc/com.atproto.server.createSession" session_data = { "identifier": username, "password": password } session_response = requests.post(session_url, json=session_data, timeout=10) session_response.raise_for_status() session = session_response.json() access_token = session.get("accessJwt") user_did = session.get("did") if not access_token or not user_did: raise Exception("Failed to get access token or DID from session") # Build reply record now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") # Create the reply reference reply_ref = { "root": {"uri": reply_to_uri, "cid": reply_to_cid}, "parent": {"uri": reply_to_uri, "cid": reply_to_cid} } post_record = { "$type": "app.bsky.feed.post", "text": text, "createdAt": now, "reply": reply_ref } # Create the post create_record_url = f"{pds_host}/xrpc/com.atproto.repo.createRecord" headers = {"Authorization": f"Bearer {access_token}"} create_data = { "repo": user_did, "collection": "app.bsky.feed.post", "record": post_record } post_response = requests.post(create_record_url, headers=headers, json=create_data, timeout=10) post_response.raise_for_status() result = post_response.json() post_uri = result.get("uri") handle = session.get("handle", username) rkey = post_uri.split("/")[-1] if post_uri else "" post_url = f"https://bsky.app/profile/{handle}/post/{rkey}" return f"Successfully posted reply to Bluesky!\nReply URL: {post_url}\nText: {text}" except Exception as e: raise Exception(f"Error replying to Bluesky post: {str(e)}")