thread

examples/atproto_mcp/README.md

@@ -13,6 +13,7 @@ This example demonstrates a FastMCP server that provides tools and resources for
 ### Tools (Actions)
 
 - **post**: Create posts with rich features (text, images, quotes, replies, links, mentions)
+- **create_thread**: Post multi-part threads with automatic linking
 - **search**: Search for posts by query
 - **follow**: Follow users by handle
 - **like**: Like posts by URI
@@ -107,6 +108,15 @@ async def demo():
             "reply_to": "at://did:plc:xxx/app.bsky.feed.post/yyy",
             "links": [{"text": "this article", "url": "https://example.com/article"}]
         })
+        
+        # Create a thread
+        await client.call_tool("create_thread", {
+            "posts": [
+                {"text": "Starting a thread about Python 🧵"},
+                {"text": "Python is great for rapid prototyping"},
+                {"text": "And the ecosystem is amazing!", "images": ["https://example.com/python.jpg"]}
+            ]
+        })
 ```
 
 ## AI Assistant Use Cases
@@ -116,7 +126,7 @@ The unified API enables natural AI assistant interactions:
 - **"Reply to that post with these findings"** → Uses `reply_to` with rich text
 - **"Share this article with commentary"** → Uses `quote` with the article link
 - **"Post this chart with explanation"** → Uses `images` with descriptive text
-- **"Start a thread about AI safety"** → Chain multiple posts with `reply_to`
+- **"Start a thread about AI safety"** → Uses `create_thread` for automatic linking
 
 ## Architecture
 

examples/atproto_mcp/demo.py

@@ -185,6 +185,37 @@ async def main(enable_posting: bool = False):
                 )
                 if json.loads(result[0].text).get("success"):
                     print("   ✅ Followed @alternatebuild.dev!")
+
+            # h. Thread creation (new!)
+            print("\n   h) Creating a thread...")
+            result = await client.call_tool(
+                "create_thread",
+                {
+                    "posts": [
+                        {
+                            "text": "Let me share some thoughts about the ATProto MCP server 🧵"
+                        },
+                        {
+                            "text": "First, it makes posting from the terminal incredibly smooth"
+                        },
+                        {
+                            "text": "The unified post API means one tool handles everything",
+                            "links": [
+                                {
+                                    "text": "everything",
+                                    "url": "https://github.com/jlowin/fastmcp",
+                                }
+                            ],
+                        },
+                        {
+                            "text": "And now with create_thread, multi-post threads are trivial!"
+                        },
+                    ]
+                },
+            )
+            if json.loads(result[0].text).get("success"):
+                thread_result = json.loads(result[0].text)
+                print(f"   ✅ Thread created with {thread_result['post_count']} posts!")
         else:
             print("\n5. Posting capabilities (not enabled):")
             print("   To test posting, run with --post flag")
@@ -206,6 +237,7 @@ async def main(enable_posting: bool = False):
         print("       • Quote posts")
         print("       • Combinations (quote + image, reply + rich text, etc.)")
         print("     - search: Search for posts")
+        print("     - create_thread: Post multi-part threads")
         print("     - follow: Follow users")
         print("     - like: Like posts")
         print("     - repost: Share posts")

examples/atproto_mcp/src/atproto_mcp/_atproto/__init__.py

@@ -1,7 +1,7 @@
 """Private ATProto implementation module."""
 
 from ._client import get_client
-from ._posts import create_post
+from ._posts import create_post, create_thread
 from ._profile import get_profile_info
 from ._read import fetch_notifications, fetch_timeline, search_for_posts
 from ._social import follow_user_by_handle, like_post_by_uri, repost_by_uri
@@ -10,6 +10,7 @@ __all__ = [
     "get_client",
     "get_profile_info",
     "create_post",
+    "create_thread",
     "fetch_timeline",
     "search_for_posts",
     "fetch_notifications",

examples/atproto_mcp/src/atproto_mcp/_atproto/_posts.py

@@ -4,7 +4,13 @@ from datetime import datetime
 
 from atproto import models
 
-from atproto_mcp.types import PostResult, RichTextLink, RichTextMention
+from atproto_mcp.types import (
+    PostResult,
+    RichTextLink,
+    RichTextMention,
+    ThreadPost,
+    ThreadResult,
+)
 
 from ._client import get_client
 
@@ -282,3 +288,90 @@ def _send_images(
         created_at=datetime.now().isoformat(),
         error=None,
     )
+
+
+def create_thread(posts: list[ThreadPost]) -> ThreadResult:
+    """Create a thread of posts with automatic linking.
+
+    Args:
+        posts: List of posts to create as a thread. First post is the root.
+    """
+    if not posts:
+        return ThreadResult(
+            success=False,
+            thread_uri=None,
+            post_uris=[],
+            post_count=0,
+            error="No posts provided",
+        )
+
+    try:
+        post_uris = []
+        root_uri = None
+        parent_uri = None
+
+        for i, post_data in enumerate(posts):
+            # First post is the root
+            if i == 0:
+                result = create_post(
+                    text=post_data["text"],
+                    images=post_data.get("images"),
+                    image_alts=post_data.get("image_alts"),
+                    links=post_data.get("links"),
+                    mentions=post_data.get("mentions"),
+                    quote=post_data.get("quote"),
+                )
+
+                if not result["success"]:
+                    return ThreadResult(
+                        success=False,
+                        thread_uri=None,
+                        post_uris=post_uris,
+                        post_count=len(post_uris),
+                        error=f"Failed to create root post: {result['error']}",
+                    )
+
+                root_uri = result["uri"]
+                parent_uri = root_uri
+                post_uris.append(root_uri)
+            else:
+                # Subsequent posts reply to the previous one
+                result = create_post(
+                    text=post_data["text"],
+                    images=post_data.get("images"),
+                    image_alts=post_data.get("image_alts"),
+                    links=post_data.get("links"),
+                    mentions=post_data.get("mentions"),
+                    quote=post_data.get("quote"),
+                    reply_to=parent_uri,
+                    reply_root=root_uri,
+                )
+
+                if not result["success"]:
+                    return ThreadResult(
+                        success=False,
+                        thread_uri=root_uri,
+                        post_uris=post_uris,
+                        post_count=len(post_uris),
+                        error=f"Failed to create post {i + 1}: {result['error']}",
+                    )
+
+                parent_uri = result["uri"]
+                post_uris.append(parent_uri)
+
+        return ThreadResult(
+            success=True,
+            thread_uri=root_uri,
+            post_uris=post_uris,
+            post_count=len(post_uris),
+            error=None,
+        )
+
+    except Exception as e:
+        return ThreadResult(
+            success=False,
+            thread_uri=None,
+            post_uris=post_uris,
+            post_count=len(post_uris),
+            error=str(e),
+        )

examples/atproto_mcp/src/atproto_mcp/server.py

@@ -16,6 +16,8 @@ from atproto_mcp.types import (
     RichTextLink,
     RichTextMention,
     SearchResult,
+    ThreadPost,
+    ThreadResult,
     TimelineResult,
 )
 from fastmcp import FastMCP
@@ -126,3 +128,27 @@ def search(
 ) -> SearchResult:
     """Search for posts containing specific text."""
     return _atproto.search_for_posts(query, limit)
+
+
+@atproto_mcp.tool
+def create_thread(
+    posts: Annotated[
+        list[ThreadPost],
+        Field(
+            description="List of posts to create as a thread. Each post can have text, images, links, mentions, and quotes."
+        ),
+    ],
+) -> ThreadResult:
+    """Create a thread of posts with automatic linking.
+
+    The first post becomes the root of the thread, and each subsequent post
+    replies to the previous one, maintaining the thread structure.
+
+    Example:
+        create_thread([
+            {"text": "Starting a thread about Python 🧵"},
+            {"text": "Python is great for rapid development"},
+            {"text": "And the ecosystem is amazing!", "images": ["https://example.com/python.jpg"]}
+        ])
+    """
+    return _atproto.create_thread(posts)

examples/atproto_mcp/src/atproto_mcp/types.py

@@ -119,3 +119,24 @@ class RichTextMention(TypedDict):
 
     handle: str
     display_text: str | None
+
+
+class ThreadPost(TypedDict, total=False):
+    """A post in a thread."""
+
+    text: str  # Required
+    images: list[str] | None
+    image_alts: list[str] | None
+    links: list[RichTextLink] | None
+    mentions: list[RichTextMention] | None
+    quote: str | None
+
+
+class ThreadResult(TypedDict):
+    """Result of creating a thread."""
+
+    success: bool
+    thread_uri: str | None  # URI of the first post
+    post_uris: list[str]
+    post_count: int
+    error: str | None