From 3264c8560653e9ddda0f9f97f60ecbf8e6adda08 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sun, 22 Jun 2025 22:29:18 -0500 Subject: [PATCH] Add advanced ATProto features: replies, rich text, quotes, and images - Add reply_to_post for threaded conversations - Add post_with_rich_text for clickable links and mentions - Add quote_post for sharing with commentary - Add post_with_images for visual content (up to 4 images) - Update types with new response structures - Add httpx dependency for image downloads - Document AI assistant use cases in README --- examples/atproto_mcp/README.md | 58 ++++- examples/atproto_mcp/pyproject.toml | 3 +- .../atproto_mcp/src/atproto_mcp/_atproto.py | 230 +++++++++++++++++- .../atproto_mcp/src/atproto_mcp/server.py | 63 +++++ examples/atproto_mcp/src/atproto_mcp/types.py | 45 ++++ 5 files changed, 390 insertions(+), 9 deletions(-) diff --git a/examples/atproto_mcp/README.md b/examples/atproto_mcp/README.md index ecc680181..1b7a7f259 100644 --- a/examples/atproto_mcp/README.md +++ b/examples/atproto_mcp/README.md @@ -15,11 +15,18 @@ The server provides two types of capabilities: ### Tools (Actions that modify state) +Basic interactions: - **post_to_bluesky**: Create new posts on Bluesky - **follow_user**: Follow a user by handle - **like_post**: Like a post by URI - **repost**: Repost content by URI +Advanced interactions: +- **reply_to_post**: Reply to posts and create threaded conversations +- **post_with_rich_text**: Create posts with clickable links and @mentions +- **quote_post**: Quote and comment on other posts +- **post_with_images**: Create posts with up to 4 images + ## Setup 1. Create a `.env` file in the root directory with your Bluesky credentials: @@ -40,7 +47,9 @@ pip install -e . python -m atproto_mcp ``` -## Usage Example +## Usage Examples + +### Basic Usage ```python from fastmcp import Client @@ -50,18 +59,53 @@ async def demo(): async with Client(atproto_mcp) as client: # Read resources status = await client.read_resource("atproto://profile/status") - print(f"Connected as: {status}") + timeline = await client.read_resource("atproto://timeline") - timeline = await client.read_resource("atproto://timeline?limit=5") - print(f"Timeline: {timeline}") - - # Use tools for actions + # Basic post post = await client.call_tool("post_to_bluesky", { "text": "Hello from FastMCP!" }) - print(f"Posted: {post}") ``` +### Advanced Usage + +```python +# Reply to a post +reply = await client.call_tool("reply_to_post", { + "parent_uri": "at://did:plc:xxx/app.bsky.feed.post/yyy", + "text": "Great point! Here's my perspective..." +}) + +# Post with rich text (links and mentions) +rich_post = await client.call_tool("post_with_rich_text", { + "text": "Check out this article by @jlowin.dev", + "links": [{"text": "this article", "url": "https://example.com"}], + "mentions": [{"handle": "jlowin.dev", "display_text": "@jlowin.dev"}] +}) + +# Quote a post +quote = await client.call_tool("quote_post", { + "text": "This is an important perspective on AI safety:", + "quoted_uri": "at://did:plc:xxx/app.bsky.feed.post/yyy" +}) + +# Post with images +image_post = await client.call_tool("post_with_images", { + "text": "Beautiful sunset today! 🌅", + "image_urls": ["https://example.com/sunset.jpg"], + "alt_texts": ["A sunset over the ocean"] +}) +``` + +## AI Assistant Use Cases + +This MCP server is designed to enable powerful AI assistant interactions: + +- **"Reply to that post about climate change with these research findings"** - Uses reply_to_post with rich text links +- **"Share this article with my thoughts"** - Uses quote_post or post_with_rich_text +- **"Post this chart with an explanation"** - Uses post_with_images +- **"Start a discussion about AI safety and mention @expert.bsky"** - Uses post_with_rich_text with mentions + ## Architecture The server is organized with: diff --git a/examples/atproto_mcp/pyproject.toml b/examples/atproto_mcp/pyproject.toml index 95abd5aa4..e874b5191 100644 --- a/examples/atproto_mcp/pyproject.toml +++ b/examples/atproto_mcp/pyproject.toml @@ -4,12 +4,13 @@ version = "0.1.0" description = "Add your description here" readme = "README.md" authors = [{ name = "zzstoatzz", email = "thrast36@gmail.com" }] -requires-python = ">=3.10" +requires-python = ">=3.12" dependencies = [ "fastmcp>=0.8.0", "atproto@git+https://github.com/MarshalX/atproto.git@refs/pull/605/head", "pydantic-settings>=2.0.0", "websockets>=15.0.1", + "httpx>=0.27.0", ] [project.scripts] diff --git a/examples/atproto_mcp/src/atproto_mcp/_atproto.py b/examples/atproto_mcp/src/atproto_mcp/_atproto.py index c13663a6b..31937a3ed 100644 --- a/examples/atproto_mcp/src/atproto_mcp/_atproto.py +++ b/examples/atproto_mcp/src/atproto_mcp/_atproto.py @@ -2,18 +2,23 @@ from datetime import datetime -from atproto import Client +from atproto import Client, models from atproto_mcp.settings import settings from atproto_mcp.types import ( FollowResult, + ImagePostResult, LikeResult, Notification, NotificationsResult, Post, PostResult, ProfileInfo, + QuotePostResult, + ReplyResult, RepostResult, + RichTextLink, + RichTextMention, SearchResult, TimelineResult, ) @@ -300,3 +305,226 @@ def repost_by_uri(uri: str) -> RepostResult: repost_uri=None, error=str(e), ) + + +def reply_to_post( + parent_uri: str, text: str, root_uri: str | None = None +) -> ReplyResult: + """Reply to a post.""" + try: + client = get_client() + + # Get parent post to extract CID + parent_post = client.app.bsky.feed.get_posts(params={"uris": [parent_uri]}) + if not parent_post.posts: + raise ValueError("Parent post not found") + + parent_cid = parent_post.posts[0].cid + parent_ref = models.create_strong_ref({"uri": parent_uri, "cid": parent_cid}) + + # If no root_uri provided, parent is the root + if root_uri is None: + root_ref = parent_ref + else: + # Get root post CID + root_post = client.app.bsky.feed.get_posts(params={"uris": [root_uri]}) + if not root_post.posts: + raise ValueError("Root post not found") + root_cid = root_post.posts[0].cid + root_ref = models.create_strong_ref({"uri": root_uri, "cid": root_cid}) + + # Create the reply + reply = client.send_post( + text=text, + reply_to=models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref), + ) + + return ReplyResult( + success=True, + uri=reply.uri, + cid=reply.cid, + parent_uri=parent_uri, + root_uri=root_uri or parent_uri, + error=None, + ) + except Exception as e: + return ReplyResult( + success=False, + uri=None, + cid=None, + parent_uri=None, + root_uri=None, + error=str(e), + ) + + +def create_post_with_rich_text( + text: str, + links: list[RichTextLink] | None = None, + mentions: list[RichTextMention] | None = None, +) -> PostResult: + """Create a post with rich text formatting (links and mentions).""" + try: + client = get_client() + facets = [] + + # Process links + if links: + for link in links: + # Find the position of link text in the main text + start = text.find(link["text"]) + if start == -1: + continue + end = start + len(link["text"]) + + facets.append( + models.AppBskyRichtextFacet.Main( + features=[models.AppBskyRichtextFacet.Link(uri=link["url"])], + index=models.AppBskyRichtextFacet.ByteSlice( + byte_start=len(text[:start].encode("UTF-8")), + byte_end=len(text[:end].encode("UTF-8")), + ), + ) + ) + + # Process mentions + if mentions: + for mention in mentions: + display_text = mention.get("display_text") or f"@{mention['handle']}" + # Find the position of mention in the main text + start = text.find(display_text) + if start == -1: + continue + end = start + len(display_text) + + # Resolve handle to DID + resolved = client.app.bsky.actor.search_actors( + params={"q": mention["handle"], "limit": 1} + ) + if not resolved.actors: + continue + + did = resolved.actors[0].did + facets.append( + models.AppBskyRichtextFacet.Main( + features=[models.AppBskyRichtextFacet.Mention(did=did)], + index=models.AppBskyRichtextFacet.ByteSlice( + byte_start=len(text[:start].encode("UTF-8")), + byte_end=len(text[:end].encode("UTF-8")), + ), + ) + ) + + # Send the post with facets + post = client.send_post(text=text, facets=facets if facets else None) + + return PostResult( + success=True, + uri=post.uri, + cid=post.cid, + text=text, + created_at=datetime.now().isoformat(), + error=None, + ) + except Exception as e: + return PostResult( + success=False, + uri=None, + cid=None, + text=None, + created_at=None, + error=str(e), + ) + + +def create_quote_post(text: str, quoted_uri: str) -> QuotePostResult: + """Create a quote post.""" + try: + client = get_client() + + # Get the post to quote + quoted_post = client.app.bsky.feed.get_posts(params={"uris": [quoted_uri]}) + if not quoted_post.posts: + raise ValueError("Quoted post not found") + + # Create strong ref for the quoted post + quoted_cid = quoted_post.posts[0].cid + quoted_ref = models.create_strong_ref({"uri": quoted_uri, "cid": quoted_cid}) + + # Create the embed + embed = models.AppBskyEmbedRecord.Main(record=quoted_ref) + + # Send the quote post + post = client.send_post(text=text, embed=embed) + + return QuotePostResult( + success=True, + uri=post.uri, + cid=post.cid, + quoted_uri=quoted_uri, + error=None, + ) + except Exception as e: + return QuotePostResult( + success=False, + uri=None, + cid=None, + quoted_uri=None, + error=str(e), + ) + + +def create_post_with_images( + text: str, + image_urls: list[str], + alt_texts: list[str] | None = None, +) -> ImagePostResult: + """Create a post with images from URLs.""" + try: + client = get_client() + import httpx + + # Ensure alt_texts has same length as images + if alt_texts is None: + alt_texts = [""] * len(image_urls) + elif len(alt_texts) < len(image_urls): + alt_texts.extend([""] * (len(image_urls) - len(alt_texts))) + + images = [] + for i, url in enumerate(image_urls[:4]): # Max 4 images + # Download image + response = httpx.get(url) + response.raise_for_status() + + # Upload to blob storage + uploaded = client.upload_blob(response.content) + + images.append( + { + "image": uploaded.blob, + "alt": alt_texts[i] if i < len(alt_texts) else "", + } + ) + + # Send post with images + post = client.send_images( + text=text, + images=[img["image"] for img in images], + image_alts=[img["alt"] for img in images], + ) + + return ImagePostResult( + success=True, + uri=post.uri, + cid=post.cid, + image_count=len(images), + error=None, + ) + except Exception as e: + return ImagePostResult( + success=False, + uri=None, + cid=None, + image_count=0, + error=str(e), + ) diff --git a/examples/atproto_mcp/src/atproto_mcp/server.py b/examples/atproto_mcp/src/atproto_mcp/server.py index 73687f27f..8341fe328 100644 --- a/examples/atproto_mcp/src/atproto_mcp/server.py +++ b/examples/atproto_mcp/src/atproto_mcp/server.py @@ -7,11 +7,16 @@ from pydantic import Field from atproto_mcp import _atproto from atproto_mcp.types import ( FollowResult, + ImagePostResult, LikeResult, NotificationsResult, PostResult, ProfileInfo, + QuotePostResult, + ReplyResult, RepostResult, + RichTextLink, + RichTextMention, SearchResult, TimelineResult, ) @@ -93,3 +98,61 @@ def repost( ) -> RepostResult: """Repost a post by its AT URI.""" return _atproto.repost_by_uri(uri) + + +# Advanced tools for richer interactions +@atproto_mcp.tool +def reply_to_post( + parent_uri: Annotated[str, Field(description="The AT URI of the post to reply to")], + text: Annotated[str, Field(max_length=300, description="The reply text")], + root_uri: Annotated[ + str | None, Field(description="The AT URI of the thread root (optional)") + ] = None, +) -> ReplyResult: + """Reply to a post, creating a threaded conversation.""" + return _atproto.reply_to_post(parent_uri, text, root_uri) + + +@atproto_mcp.tool +def post_with_rich_text( + text: Annotated[ + str, + Field( + max_length=300, + description="The post text with placeholders for links/mentions", + ), + ], + links: Annotated[ + list[RichTextLink] | None, Field(description="Links to embed in the text") + ] = None, + mentions: Annotated[ + list[RichTextMention] | None, Field(description="User mentions to embed") + ] = None, +) -> PostResult: + """Create a post with rich text formatting including clickable links and mentions.""" + return _atproto.create_post_with_rich_text(text, links, mentions) + + +@atproto_mcp.tool +def quote_post( + text: Annotated[ + str, Field(max_length=300, description="Your commentary on the quoted post") + ], + quoted_uri: Annotated[str, Field(description="The AT URI of the post to quote")], +) -> QuotePostResult: + """Create a quote post to share and comment on another post.""" + return _atproto.create_quote_post(text, quoted_uri) + + +@atproto_mcp.tool +def post_with_images( + text: Annotated[str, Field(max_length=300, description="The post text")], + image_urls: Annotated[ + list[str], Field(max_length=4, description="URLs of images to attach (max 4)") + ], + alt_texts: Annotated[ + list[str] | None, Field(description="Alt text for each image") + ] = None, +) -> ImagePostResult: + """Create a post with images attached.""" + return _atproto.create_post_with_images(text, image_urls, alt_texts) diff --git a/examples/atproto_mcp/src/atproto_mcp/types.py b/examples/atproto_mcp/src/atproto_mcp/types.py index 4cf726448..d452dcb84 100644 --- a/examples/atproto_mcp/src/atproto_mcp/types.py +++ b/examples/atproto_mcp/src/atproto_mcp/types.py @@ -103,3 +103,48 @@ class RepostResult(TypedDict): reposted_uri: str | None repost_uri: str | None error: str | None + + +class ReplyResult(TypedDict): + """Result of replying to a post.""" + + success: bool + uri: str | None + cid: str | None + parent_uri: str | None + root_uri: str | None + error: str | None + + +class RichTextLink(TypedDict): + """A link in rich text.""" + + text: str + url: str + + +class RichTextMention(TypedDict): + """A mention in rich text.""" + + handle: str + display_text: str | None + + +class QuotePostResult(TypedDict): + """Result of creating a quote post.""" + + success: bool + uri: str | None + cid: str | None + quoted_uri: str | None + error: str | None + + +class ImagePostResult(TypedDict): + """Result of posting with images.""" + + success: bool + uri: str | None + cid: str | None + image_count: int + error: str | None