Merge pull request #916 from jlowin/atproto-example

Add ATProto (Bluesky) MCP Server Example

examples/atproto_mcp/README.md

@@ -0,0 +1,146 @@
+# ATProto MCP Server
+
+This example demonstrates a FastMCP server that provides tools and resources for interacting with the AT Protocol (Bluesky).
+
+## Features
+
+### Resources (Read-only)
+
+- **atproto://profile/status**: Get connection status and profile information
+- **atproto://timeline**: Retrieve your timeline feed
+- **atproto://notifications**: Get recent notifications
+
+### Tools (Actions)
+
+- **post**: Create posts with rich features (text, images, quotes, replies, links, mentions)
+- **search**: Search for posts by query
+- **follow**: Follow users by handle
+- **like**: Like posts by URI
+- **repost**: Share posts by URI
+
+## Setup
+
+1. Create a `.env` file in the root directory with your Bluesky credentials:
+
+```bash
+ATPROTO_HANDLE=your.handle@bsky.social
+ATPROTO_PASSWORD=your-app-password
+ATPROTO_PDS_URL=https://bsky.social  # optional, defaults to bsky.social
+```
+
+2. Install and run the server:
+
+```bash
+# Install dependencies
+uv pip install -e .
+
+# Run the server
+uv run atproto-mcp
+```
+
+## The Unified Post Tool
+
+The `post` tool is a single, flexible interface for all posting needs:
+
+```python
+async def post(
+    text: str,                          # Required: Post content
+    images: list[str] = None,           # Optional: Image URLs (max 4)
+    image_alts: list[str] = None,       # Optional: Alt text for images
+    links: list[RichTextLink] = None,   # Optional: Embedded links
+    mentions: list[RichTextMention] = None,  # Optional: User mentions
+    reply_to: str = None,               # Optional: Reply to post URI
+    reply_root: str = None,             # Optional: Thread root URI
+    quote: str = None,                  # Optional: Quote post URI
+)
+```
+
+### Usage Examples
+
+```python
+from fastmcp import Client
+from atproto_mcp.server import atproto_mcp
+
+async def demo():
+    async with Client(atproto_mcp) as client:
+        # Simple post
+        await client.call_tool("post", {
+            "text": "Hello from FastMCP!"
+        })
+        
+        # Post with image
+        await client.call_tool("post", {
+            "text": "Beautiful sunset! 🌅",
+            "images": ["https://example.com/sunset.jpg"],
+            "image_alts": ["Sunset over the ocean"]
+        })
+        
+        # Reply to a post
+        await client.call_tool("post", {
+            "text": "Great point!",
+            "reply_to": "at://did:plc:xxx/app.bsky.feed.post/yyy"
+        })
+        
+        # Quote post
+        await client.call_tool("post", {
+            "text": "This is important:",
+            "quote": "at://did:plc:xxx/app.bsky.feed.post/yyy"
+        })
+        
+        # Rich text with links and mentions
+        await client.call_tool("post", {
+            "text": "Check out FastMCP by @alternatebuild.dev",
+            "links": [{"text": "FastMCP", "url": "https://github.com/jlowin/fastmcp"}],
+            "mentions": [{"handle": "alternatebuild.dev", "display_text": "@alternatebuild.dev"}]
+        })
+        
+        # Advanced: Quote with image
+        await client.call_tool("post", {
+            "text": "Adding visual context:",
+            "quote": "at://did:plc:xxx/app.bsky.feed.post/yyy",
+            "images": ["https://example.com/chart.png"]
+        })
+        
+        # Advanced: Reply with rich text
+        await client.call_tool("post", {
+            "text": "I agree! See this article for more info",
+            "reply_to": "at://did:plc:xxx/app.bsky.feed.post/yyy",
+            "links": [{"text": "this article", "url": "https://example.com/article"}]
+        })
+```
+
+## AI Assistant Use Cases
+
+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`
+
+## Architecture
+
+The server is organized as:
+- `server.py` - Public API with resources and tools
+- `_atproto/` - Private implementation module
+  - `_client.py` - ATProto client management
+  - `_posts.py` - Unified posting logic
+  - `_profile.py` - Profile operations
+  - `_read.py` - Timeline, search, notifications
+  - `_social.py` - Follow, like, repost
+- `types.py` - TypedDict definitions
+- `settings.py` - Configuration management
+
+## Running the Demo
+
+```bash
+# Run demo (read-only)
+uv run python demo.py
+
+# Run demo with posting enabled
+uv run python demo.py --post
+```
+
+## Security Note
+
+Store your Bluesky credentials securely in environment variables. Never commit credentials to version control.

examples/atproto_mcp/demo.py

@@ -0,0 +1,225 @@
+"""Demo script showing all ATProto MCP server capabilities."""
+
+import argparse
+import asyncio
+import json
+from typing import cast
+
+from atproto_mcp.server import atproto_mcp
+from atproto_mcp.types import (
+    NotificationsResult,
+    PostResult,
+    ProfileInfo,
+    SearchResult,
+    TimelineResult,
+)
+
+from fastmcp import Client
+
+
+async def main(enable_posting: bool = False):
+    print("🔵 ATProto MCP Server Demo\n")
+
+    async with Client(atproto_mcp) as client:
+        # 1. Check connection status (resource)
+        print("1. Checking connection status...")
+        result = await client.read_resource("atproto://profile/status")
+        status: ProfileInfo = (
+            json.loads(result[0].text) if result else cast(ProfileInfo, {})
+        )
+
+        if status.get("connected"):
+            print(f"✅ Connected as: @{status['handle']}")
+            print(f"   Followers: {status['followers']}")
+            print(f"   Following: {status['following']}")
+            print(f"   Posts: {status['posts']}")
+        else:
+            print(f"❌ Connection failed: {status.get('error')}")
+            return
+
+        # 2. Get timeline
+        print("\n2. Getting timeline...")
+        result = await client.read_resource("atproto://timeline")
+        timeline: TimelineResult = (
+            json.loads(result[0].text) if result else cast(TimelineResult, {})
+        )
+
+        if timeline.get("success") and timeline["posts"]:
+            print(f"✅ Found {timeline['count']} posts")
+            post = timeline["posts"][0]
+            print(f"   Latest by @{post['author']}: {post['text'][:80]}...")
+            save_uri = post["uri"]  # Save for later interactions
+        else:
+            print("❌ No posts in timeline")
+            save_uri = None
+
+        # 3. Search for posts
+        print("\n3. Searching for posts about 'Bluesky'...")
+        result = await client.call_tool("search", {"query": "Bluesky", "limit": 5})
+        search: SearchResult = (
+            json.loads(result[0].text) if result else cast(SearchResult, {})
+        )
+
+        if search.get("success") and search["posts"]:
+            print(f"✅ Found {search['count']} posts")
+            print(f"   Sample: {search['posts'][0]['text'][:80]}...")
+
+        # 4. Get notifications
+        print("\n4. Checking notifications...")
+        result = await client.read_resource("atproto://notifications")
+        notifs: NotificationsResult = (
+            json.loads(result[0].text) if result else cast(NotificationsResult, {})
+        )
+
+        if notifs.get("success"):
+            print(f"✅ You have {notifs['count']} notifications")
+            unread = sum(1 for n in notifs["notifications"] if not n["is_read"])
+            if unread:
+                print(f"   ({unread} unread)")
+
+        # 5. Demo posting capabilities
+        if enable_posting:
+            print("\n5. Demonstrating posting capabilities...")
+
+            # a. Simple post
+            print("\n   a) Creating a simple post...")
+            result = await client.call_tool(
+                "post",
+                {"text": "🧪 Testing the unified ATProto MCP post tool! #FastMCP"},
+            )
+            post_result: PostResult = json.loads(result[0].text) if result else {}
+            if post_result.get("success"):
+                print("   ✅ Posted successfully!")
+                simple_uri = post_result["uri"]
+            else:
+                print(f"   ❌ Failed: {post_result.get('error')}")
+                simple_uri = None
+
+            # b. Post with rich text (link and mention)
+            print("\n   b) Creating a post with rich text...")
+            result = await client.call_tool(
+                "post",
+                {
+                    "text": "Check out FastMCP and follow @alternatebuild.dev for updates!",
+                    "links": [
+                        {"text": "FastMCP", "url": "https://github.com/jlowin/fastmcp"}
+                    ],
+                    "mentions": [
+                        {
+                            "handle": "alternatebuild.dev",
+                            "display_text": "@alternatebuild.dev",
+                        }
+                    ],
+                },
+            )
+            if json.loads(result[0].text).get("success"):
+                print("   ✅ Rich text post created!")
+
+            # c. Reply to a post
+            if save_uri:
+                print("\n   c) Replying to a post...")
+                result = await client.call_tool(
+                    "post", {"text": "Great post! 👍", "reply_to": save_uri}
+                )
+                if json.loads(result[0].text).get("success"):
+                    print("   ✅ Reply posted!")
+
+            # d. Quote post
+            if simple_uri:
+                print("\n   d) Creating a quote post...")
+                result = await client.call_tool(
+                    "post",
+                    {
+                        "text": "Quoting my own test post for demo purposes 🔄",
+                        "quote": simple_uri,
+                    },
+                )
+                if json.loads(result[0].text).get("success"):
+                    print("   ✅ Quote post created!")
+
+            # e. Post with image
+            print("\n   e) Creating a post with image...")
+            result = await client.call_tool(
+                "post",
+                {
+                    "text": "Here's a test image post! 📸",
+                    "images": ["https://picsum.photos/400/300"],
+                    "image_alts": ["Random test image"],
+                },
+            )
+            if json.loads(result[0].text).get("success"):
+                print("   ✅ Image post created!")
+
+            # f. Quote with image (advanced)
+            if simple_uri:
+                print("\n   f) Creating a quote post with image...")
+                result = await client.call_tool(
+                    "post",
+                    {
+                        "text": "Quote + image combo! 🎨",
+                        "quote": simple_uri,
+                        "images": ["https://picsum.photos/300/200"],
+                        "image_alts": ["Another test image"],
+                    },
+                )
+                if json.loads(result[0].text).get("success"):
+                    print("   ✅ Quote with image created!")
+
+            # g. Social actions
+            if save_uri:
+                print("\n   g) Demonstrating social actions...")
+
+                # Like
+                result = await client.call_tool("like", {"uri": save_uri})
+                if json.loads(result[0].text).get("success"):
+                    print("   ✅ Liked a post!")
+
+                # Repost
+                result = await client.call_tool("repost", {"uri": save_uri})
+                if json.loads(result[0].text).get("success"):
+                    print("   ✅ Reposted!")
+
+                # Follow
+                result = await client.call_tool(
+                    "follow", {"handle": "alternatebuild.dev"}
+                )
+                if json.loads(result[0].text).get("success"):
+                    print("   ✅ Followed @alternatebuild.dev!")
+        else:
+            print("\n5. Posting capabilities (not enabled):")
+            print("   To test posting, run with --post flag")
+            print("   Example: python demo.py --post")
+
+        # 6. Show available capabilities
+        print("\n6. Available capabilities:")
+        print("\n   Resources (read-only):")
+        print("     - atproto://profile/status")
+        print("     - atproto://timeline")
+        print("     - atproto://notifications")
+
+        print("\n   Tools (actions):")
+        print("     - post: Unified posting with rich features")
+        print("       • Simple text posts")
+        print("       • Images (up to 4)")
+        print("       • Rich text (links, mentions)")
+        print("       • Replies and threads")
+        print("       • Quote posts")
+        print("       • Combinations (quote + image, reply + rich text, etc.)")
+        print("     - search: Search for posts")
+        print("     - follow: Follow users")
+        print("     - like: Like posts")
+        print("     - repost: Share posts")
+
+        print("\n✨ Demo complete!")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="ATProto MCP Server Demo")
+    parser.add_argument(
+        "--post",
+        action="store_true",
+        help="Enable posting test messages to Bluesky",
+    )
+    args = parser.parse_args()
+
+    asyncio.run(main(enable_posting=args.post))

examples/atproto_mcp/pyproject.toml

@@ -0,0 +1,27 @@
+[project]
+name = "atproto-mcp"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+authors = [{ name = "zzstoatzz", email = "thrast36@gmail.com" }]
+requires-python = ">=3.10"
+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]
+atproto-mcp = "atproto_mcp.__main__:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.metadata]
+allow-direct-references = true
+
+[tool.uv.sources]
+fastmcp = { workspace = true }

examples/atproto_mcp/src/atproto_mcp/__init__.py

@@ -0,0 +1,3 @@
+from atproto_mcp.settings import settings
+
+__all__ = ["settings"]

examples/atproto_mcp/src/atproto_mcp/__main__.py

@@ -0,0 +1,9 @@
+from atproto_mcp.server import atproto_mcp
+
+
+def main():
+    atproto_mcp.run()
+
+
+if __name__ == "__main__":
+    main()

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

@@ -0,0 +1,19 @@
+"""Private ATProto implementation module."""
+
+from ._client import get_client
+from ._posts import create_post
+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
+
+__all__ = [
+    "get_client",
+    "get_profile_info",
+    "create_post",
+    "fetch_timeline",
+    "search_for_posts",
+    "fetch_notifications",
+    "follow_user_by_handle",
+    "like_post_by_uri",
+    "repost_by_uri",
+]

examples/atproto_mcp/src/atproto_mcp/_atproto/_client.py

@@ -0,0 +1,16 @@
+"""ATProto client management."""
+
+from atproto import Client
+
+from atproto_mcp.settings import settings
+
+_client: Client | None = None
+
+
+def get_client() -> Client:
+    """Get or create an authenticated ATProto client."""
+    global _client
+    if _client is None:
+        _client = Client()
+        _client.login(settings.atproto_handle, settings.atproto_password)
+    return _client

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

@@ -0,0 +1,284 @@
+"""Unified posting functionality."""
+
+from datetime import datetime
+
+from atproto import models
+
+from atproto_mcp.types import PostResult, RichTextLink, RichTextMention
+
+from ._client import get_client
+
+
+def create_post(
+    text: str,
+    images: list[str] | None = None,
+    image_alts: list[str] | None = None,
+    links: list[RichTextLink] | None = None,
+    mentions: list[RichTextMention] | None = None,
+    reply_to: str | None = None,
+    reply_root: str | None = None,
+    quote: str | None = None,
+) -> PostResult:
+    """Create a unified post with optional features.
+
+    Args:
+        text: Post text (max 300 chars)
+        images: URLs of images to attach (max 4)
+        image_alts: Alt text for images
+        links: Links to embed in rich text
+        mentions: User mentions to embed
+        reply_to: URI of post to reply to
+        reply_root: URI of thread root (defaults to reply_to)
+        quote: URI of post to quote
+    """
+    try:
+        client = get_client()
+        facets = []
+        embed = None
+        reply_ref = None
+
+        # Handle rich text facets (links and mentions)
+        if links or mentions:
+            facets = _build_facets(text, links, mentions, client)
+
+        # Handle replies
+        if reply_to:
+            reply_ref = _build_reply_ref(reply_to, reply_root, client)
+
+        # Handle quotes and images
+        if quote and images:
+            # Quote with images - create record with media embed
+            embed = _build_quote_with_images_embed(quote, images, image_alts, client)
+        elif quote:
+            # Quote only
+            embed = _build_quote_embed(quote, client)
+        elif images:
+            # Images only - use send_images for proper handling
+            return _send_images(text, images, image_alts, facets, reply_ref, client)
+
+        # Send the post
+        post = client.send_post(
+            text=text,
+            facets=facets if facets else None,
+            embed=embed,
+            reply_to=reply_ref,
+        )
+
+        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 _build_facets(
+    text: str,
+    links: list[RichTextLink] | None,
+    mentions: list[RichTextMention] | None,
+    client,
+):
+    """Build facets for rich text formatting."""
+    facets = []
+
+    # Process links
+    if links:
+        for link in links:
+            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']}"
+            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")),
+                    ),
+                )
+            )
+
+    return facets
+
+
+def _build_reply_ref(reply_to: str, reply_root: str | None, client):
+    """Build reply reference."""
+    # Get parent post to extract CID
+    parent_post = client.app.bsky.feed.get_posts(params={"uris": [reply_to]})
+    if not parent_post.posts:
+        raise ValueError("Parent post not found")
+
+    parent_cid = parent_post.posts[0].cid
+    parent_ref = models.ComAtprotoRepoStrongRef.Main(uri=reply_to, cid=parent_cid)
+
+    # If no root_uri provided, parent is the root
+    if reply_root is None:
+        root_ref = parent_ref
+    else:
+        # Get root post CID
+        root_post = client.app.bsky.feed.get_posts(params={"uris": [reply_root]})
+        if not root_post.posts:
+            raise ValueError("Root post not found")
+        root_cid = root_post.posts[0].cid
+        root_ref = models.ComAtprotoRepoStrongRef.Main(uri=reply_root, cid=root_cid)
+
+    return models.AppBskyFeedPost.ReplyRef(parent=parent_ref, root=root_ref)
+
+
+def _build_quote_embed(quote_uri: str, client):
+    """Build quote embed."""
+    # Get the post to quote
+    quoted_post = client.app.bsky.feed.get_posts(params={"uris": [quote_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.ComAtprotoRepoStrongRef.Main(uri=quote_uri, cid=quoted_cid)
+
+    # Create the embed
+    return models.AppBskyEmbedRecord.Main(record=quoted_ref)
+
+
+def _build_quote_with_images_embed(
+    quote_uri: str, image_urls: list[str], image_alts: list[str] | None, client
+):
+    """Build quote embed with images."""
+    import httpx
+
+    # Get the quoted post
+    quoted_post = client.app.bsky.feed.get_posts(params={"uris": [quote_uri]})
+    if not quoted_post.posts:
+        raise ValueError("Quoted post not found")
+
+    quoted_cid = quoted_post.posts[0].cid
+    quoted_ref = models.ComAtprotoRepoStrongRef.Main(uri=quote_uri, cid=quoted_cid)
+
+    # Download and upload images
+    images = []
+    alts = image_alts or [""] * len(image_urls)
+
+    for i, url in enumerate(image_urls[:4]):
+        response = httpx.get(url, follow_redirects=True)
+        response.raise_for_status()
+
+        # Upload to blob storage
+        upload = client.upload_blob(response.content)
+        images.append(
+            models.AppBskyEmbedImages.Image(
+                alt=alts[i] if i < len(alts) else "",
+                image=upload.blob,
+            )
+        )
+
+    # Create record with media embed
+    return models.AppBskyEmbedRecordWithMedia.Main(
+        record=models.AppBskyEmbedRecord.Main(record=quoted_ref),
+        media=models.AppBskyEmbedImages.Main(images=images),
+    )
+
+
+def _send_images(
+    text: str,
+    image_urls: list[str],
+    image_alts: list[str] | None,
+    facets,
+    reply_ref,
+    client,
+):
+    """Send post with images using the client's send_images method."""
+    import httpx
+
+    # Ensure alt_texts has same length as images
+    if image_alts is None:
+        image_alts = [""] * len(image_urls)
+    elif len(image_alts) < len(image_urls):
+        image_alts.extend([""] * (len(image_urls) - len(image_alts)))
+
+    image_data = []
+    alts = []
+    for i, url in enumerate(image_urls[:4]):  # Max 4 images
+        # Download image (follow redirects)
+        response = httpx.get(url, follow_redirects=True)
+        response.raise_for_status()
+
+        image_data.append(response.content)
+        alts.append(image_alts[i] if i < len(image_alts) else "")
+
+    # Send post with images
+    # Note: send_images doesn't support facets or reply_to directly
+    # So we need to use send_post with manual image upload if we have those
+    if facets or reply_ref:
+        # Manual image upload
+        images = []
+        for i, data in enumerate(image_data):
+            upload = client.upload_blob(data)
+            images.append(
+                models.AppBskyEmbedImages.Image(
+                    alt=alts[i],
+                    image=upload.blob,
+                )
+            )
+
+        embed = models.AppBskyEmbedImages.Main(images=images)
+        post = client.send_post(
+            text=text,
+            facets=facets if facets else None,
+            embed=embed,
+            reply_to=reply_ref,
+        )
+    else:
+        # Use simple send_images
+        post = client.send_images(
+            text=text,
+            images=image_data,
+            image_alts=alts,
+        )
+
+    return PostResult(
+        success=True,
+        uri=post.uri,
+        cid=post.cid,
+        text=text,
+        created_at=datetime.now().isoformat(),
+        error=None,
+    )

examples/atproto_mcp/src/atproto_mcp/_atproto/_profile.py

@@ -0,0 +1,33 @@
+"""Profile-related operations."""
+
+from atproto_mcp.types import ProfileInfo
+
+from ._client import get_client
+
+
+def get_profile_info() -> ProfileInfo:
+    """Get profile information for the authenticated user."""
+    try:
+        client = get_client()
+        profile = client.get_profile(client.me.did)
+        return ProfileInfo(
+            connected=True,
+            handle=profile.handle,
+            display_name=profile.display_name,
+            did=client.me.did,
+            followers=profile.followers_count,
+            following=profile.follows_count,
+            posts=profile.posts_count,
+            error=None,
+        )
+    except Exception as e:
+        return ProfileInfo(
+            connected=False,
+            handle=None,
+            display_name=None,
+            did=None,
+            followers=None,
+            following=None,
+            posts=None,
+            error=str(e),
+        )

examples/atproto_mcp/src/atproto_mcp/_atproto/_read.py

@@ -0,0 +1,124 @@
+"""Read-only operations for timeline, search, and notifications."""
+
+from atproto_mcp.types import (
+    Notification,
+    NotificationsResult,
+    Post,
+    SearchResult,
+    TimelineResult,
+)
+
+from ._client import get_client
+
+
+def fetch_timeline(limit: int = 10) -> TimelineResult:
+    """Fetch the authenticated user's timeline."""
+    try:
+        client = get_client()
+        timeline = client.get_timeline(limit=limit)
+
+        posts = []
+        for feed_view in timeline.feed:
+            post = feed_view.post
+            posts.append(
+                Post(
+                    uri=post.uri,
+                    cid=post.cid,
+                    text=post.record.text if hasattr(post.record, "text") else "",
+                    author=post.author.handle,
+                    created_at=post.record.created_at,
+                    likes=post.like_count or 0,
+                    reposts=post.repost_count or 0,
+                    replies=post.reply_count or 0,
+                )
+            )
+
+        return TimelineResult(
+            success=True,
+            posts=posts,
+            count=len(posts),
+            error=None,
+        )
+    except Exception as e:
+        return TimelineResult(
+            success=False,
+            posts=[],
+            count=0,
+            error=str(e),
+        )
+
+
+def search_for_posts(query: str, limit: int = 10) -> SearchResult:
+    """Search for posts containing specific text."""
+    try:
+        client = get_client()
+        search_results = client.app.bsky.feed.search_posts(
+            params={"q": query, "limit": limit}
+        )
+
+        posts = []
+        for post in search_results.posts:
+            posts.append(
+                Post(
+                    uri=post.uri,
+                    cid=post.cid,
+                    text=post.record.text if hasattr(post.record, "text") else "",
+                    author=post.author.handle,
+                    created_at=post.record.created_at,
+                    likes=post.like_count or 0,
+                    reposts=post.repost_count or 0,
+                    replies=post.reply_count or 0,
+                )
+            )
+
+        return SearchResult(
+            success=True,
+            query=query,
+            posts=posts,
+            count=len(posts),
+            error=None,
+        )
+    except Exception as e:
+        return SearchResult(
+            success=False,
+            query=query,
+            posts=[],
+            count=0,
+            error=str(e),
+        )
+
+
+def fetch_notifications(limit: int = 10) -> NotificationsResult:
+    """Fetch recent notifications."""
+    try:
+        client = get_client()
+        notifs = client.app.bsky.notification.list_notifications(
+            params={"limit": limit}
+        )
+
+        notifications = []
+        for notif in notifs.notifications:
+            notifications.append(
+                Notification(
+                    uri=notif.uri,
+                    cid=notif.cid,
+                    author=notif.author.handle,
+                    reason=notif.reason,
+                    is_read=notif.is_read,
+                    indexed_at=notif.indexed_at,
+                )
+            )
+
+        return NotificationsResult(
+            success=True,
+            notifications=notifications,
+            count=len(notifications),
+            error=None,
+        )
+    except Exception as e:
+        return NotificationsResult(
+            success=False,
+            notifications=[],
+            count=0,
+            error=str(e),
+        )

examples/atproto_mcp/src/atproto_mcp/_atproto/_social.py

@@ -0,0 +1,108 @@
+"""Social actions like follow, like, and repost."""
+
+from atproto_mcp.types import FollowResult, LikeResult, RepostResult
+
+from ._client import get_client
+
+
+def follow_user_by_handle(handle: str) -> FollowResult:
+    """Follow a user by their handle."""
+    try:
+        client = get_client()
+        # Search for the user to get their DID
+        results = client.app.bsky.actor.search_actors(params={"q": handle, "limit": 1})
+        if not results.actors:
+            return FollowResult(
+                success=False,
+                did=None,
+                handle=None,
+                uri=None,
+                error=f"User @{handle} not found",
+            )
+
+        actor = results.actors[0]
+        # Create the follow
+        follow = client.follow(actor.did)
+        return FollowResult(
+            success=True,
+            did=actor.did,
+            handle=actor.handle,
+            uri=follow.uri,
+            error=None,
+        )
+    except Exception as e:
+        return FollowResult(
+            success=False,
+            did=None,
+            handle=None,
+            uri=None,
+            error=str(e),
+        )
+
+
+def like_post_by_uri(uri: str) -> LikeResult:
+    """Like a post by its AT URI."""
+    try:
+        client = get_client()
+        # Parse the URI to get the components
+        # URI format: at://did:plc:xxx/app.bsky.feed.post/yyy
+        parts = uri.replace("at://", "").split("/")
+        if len(parts) != 3 or parts[1] != "app.bsky.feed.post":
+            raise ValueError("Invalid post URI format")
+
+        # Get the post to retrieve its CID
+        post = client.app.bsky.feed.get_posts(params={"uris": [uri]})
+        if not post.posts:
+            raise ValueError("Post not found")
+
+        cid = post.posts[0].cid
+
+        # Now like the post with both URI and CID
+        like = client.like(uri, cid)
+        return LikeResult(
+            success=True,
+            liked_uri=uri,
+            like_uri=like.uri,
+            error=None,
+        )
+    except Exception as e:
+        return LikeResult(
+            success=False,
+            liked_uri=None,
+            like_uri=None,
+            error=str(e),
+        )
+
+
+def repost_by_uri(uri: str) -> RepostResult:
+    """Repost a post by its AT URI."""
+    try:
+        client = get_client()
+        # Parse the URI to get the components
+        # URI format: at://did:plc:xxx/app.bsky.feed.post/yyy
+        parts = uri.replace("at://", "").split("/")
+        if len(parts) != 3 or parts[1] != "app.bsky.feed.post":
+            raise ValueError("Invalid post URI format")
+
+        # Get the post to retrieve its CID
+        post = client.app.bsky.feed.get_posts(params={"uris": [uri]})
+        if not post.posts:
+            raise ValueError("Post not found")
+
+        cid = post.posts[0].cid
+
+        # Now repost with both URI and CID
+        repost = client.repost(uri, cid)
+        return RepostResult(
+            success=True,
+            reposted_uri=uri,
+            repost_uri=repost.uri,
+            error=None,
+        )
+    except Exception as e:
+        return RepostResult(
+            success=False,
+            reposted_uri=None,
+            repost_uri=None,
+            error=str(e),
+        )

examples/atproto_mcp/src/atproto_mcp/server.py

@@ -0,0 +1,128 @@
+"""ATProto MCP Server - Public API exposing Bluesky tools and resources."""
+
+from typing import Annotated
+
+from pydantic import Field
+
+from atproto_mcp import _atproto
+from atproto_mcp.settings import settings
+from atproto_mcp.types import (
+    FollowResult,
+    LikeResult,
+    NotificationsResult,
+    PostResult,
+    ProfileInfo,
+    RepostResult,
+    RichTextLink,
+    RichTextMention,
+    SearchResult,
+    TimelineResult,
+)
+from fastmcp import FastMCP
+
+atproto_mcp = FastMCP(
+    "ATProto MCP Server",
+    dependencies=[
+        "atproto_mcp@git+https://github.com/jlowin/fastmcp.git#subdirectory=examples/atproto_mcp",
+    ],
+)
+
+
+# Resources - read-only operations
+@atproto_mcp.resource("atproto://profile/status")
+def atproto_status() -> ProfileInfo:
+    """Check the status of the ATProto connection and current user profile."""
+    return _atproto.get_profile_info()
+
+
+@atproto_mcp.resource("atproto://timeline")
+def get_timeline() -> TimelineResult:
+    """Get the authenticated user's timeline feed."""
+    return _atproto.fetch_timeline(settings.atproto_timeline_default_limit)
+
+
+@atproto_mcp.resource("atproto://notifications")
+def get_notifications() -> NotificationsResult:
+    """Get recent notifications for the authenticated user."""
+    return _atproto.fetch_notifications(settings.atproto_notifications_default_limit)
+
+
+# Tools - actions that modify state
+@atproto_mcp.tool
+def post(
+    text: Annotated[
+        str, Field(max_length=300, description="The text content of the post")
+    ],
+    images: Annotated[
+        list[str] | None,
+        Field(max_length=4, description="URLs of images to attach (max 4)"),
+    ] = None,
+    image_alts: Annotated[
+        list[str] | None, Field(description="Alt text for each image")
+    ] = None,
+    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,
+    reply_to: Annotated[
+        str | None, Field(description="AT URI of post to reply to")
+    ] = None,
+    reply_root: Annotated[
+        str | None, Field(description="AT URI of thread root (defaults to reply_to)")
+    ] = None,
+    quote: Annotated[str | None, Field(description="AT URI of post to quote")] = None,
+) -> PostResult:
+    """Create a post with optional rich features like images, quotes, replies, and rich text.
+
+    Examples:
+        - Simple post: post("Hello world!")
+        - With image: post("Check this out!", images=["https://example.com/img.jpg"])
+        - Reply: post("I agree!", reply_to="at://did/app.bsky.feed.post/123")
+        - Quote: post("Great point!", quote="at://did/app.bsky.feed.post/456")
+        - Rich text: post("Check out example.com", links=[{"text": "example.com", "url": "https://example.com"}])
+    """
+    return _atproto.create_post(
+        text, images, image_alts, links, mentions, reply_to, reply_root, quote
+    )
+
+
+@atproto_mcp.tool
+def follow(
+    handle: Annotated[
+        str,
+        Field(
+            description="The handle of the user to follow (e.g., 'user.bsky.social')"
+        ),
+    ],
+) -> FollowResult:
+    """Follow a user by their handle."""
+    return _atproto.follow_user_by_handle(handle)
+
+
+@atproto_mcp.tool
+def like(
+    uri: Annotated[str, Field(description="The AT URI of the post to like")],
+) -> LikeResult:
+    """Like a post by its AT URI."""
+    return _atproto.like_post_by_uri(uri)
+
+
+@atproto_mcp.tool
+def repost(
+    uri: Annotated[str, Field(description="The AT URI of the post to repost")],
+) -> RepostResult:
+    """Repost a post by its AT URI."""
+    return _atproto.repost_by_uri(uri)
+
+
+@atproto_mcp.tool
+def search(
+    query: Annotated[str, Field(description="Search query for posts")],
+    limit: Annotated[
+        int, Field(ge=1, le=100, description="Number of results to return")
+    ] = settings.atproto_search_default_limit,
+) -> SearchResult:
+    """Search for posts containing specific text."""
+    return _atproto.search_for_posts(query, limit)

examples/atproto_mcp/src/atproto_mcp/settings.py

@@ -0,0 +1,17 @@
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(env_file=[".env"], extra="ignore")
+
+    atproto_handle: str = Field(default=...)
+    atproto_password: str = Field(default=...)
+    atproto_pds_url: str = Field(default="https://bsky.social")
+
+    atproto_notifications_default_limit: int = Field(default=10)
+    atproto_timeline_default_limit: int = Field(default=10)
+    atproto_search_default_limit: int = Field(default=10)
+
+
+settings = Settings()

examples/atproto_mcp/src/atproto_mcp/types.py

@@ -0,0 +1,121 @@
+"""Type definitions for ATProto MCP server."""
+
+from typing import TypedDict
+
+
+class ProfileInfo(TypedDict):
+    """Profile information response."""
+
+    connected: bool
+    handle: str | None
+    display_name: str | None
+    did: str | None
+    followers: int | None
+    following: int | None
+    posts: int | None
+    error: str | None
+
+
+class PostResult(TypedDict):
+    """Result of creating a post."""
+
+    success: bool
+    uri: str | None
+    cid: str | None
+    text: str | None
+    created_at: str | None
+    error: str | None
+
+
+class Post(TypedDict):
+    """A single post."""
+
+    author: str
+    text: str | None
+    created_at: str | None
+    likes: int
+    reposts: int
+    replies: int
+    uri: str
+    cid: str
+
+
+class TimelineResult(TypedDict):
+    """Timeline fetch result."""
+
+    success: bool
+    count: int
+    posts: list[Post]
+    error: str | None
+
+
+class SearchResult(TypedDict):
+    """Search result."""
+
+    success: bool
+    query: str
+    count: int
+    posts: list[Post]
+    error: str | None
+
+
+class Notification(TypedDict):
+    """A single notification."""
+
+    reason: str
+    author: str | None
+    is_read: bool
+    indexed_at: str
+    uri: str
+    cid: str
+
+
+class NotificationsResult(TypedDict):
+    """Notifications fetch result."""
+
+    success: bool
+    count: int
+    notifications: list[Notification]
+    error: str | None
+
+
+class FollowResult(TypedDict):
+    """Result of following a user."""
+
+    success: bool
+    handle: str | None
+    did: str | None
+    uri: str | None
+    error: str | None
+
+
+class LikeResult(TypedDict):
+    """Result of liking a post."""
+
+    success: bool
+    liked_uri: str | None
+    like_uri: str | None
+    error: str | None
+
+
+class RepostResult(TypedDict):
+    """Result of reposting."""
+
+    success: bool
+    reposted_uri: str | None
+    repost_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