Implement full ATProto functionality with 8 tools for Bluesky interaction

examples/atproto_mcp/README.md

@@ -0,0 +1,70 @@
+# ATProto MCP Server
+
+This example demonstrates a FastMCP server that provides tools for interacting with the AT Protocol (Bluesky).
+
+## Features
+
+The server provides the following tools:
+
+- **atproto_status**: Check connection status and profile information
+- **post_to_bluesky**: Create new posts on Bluesky
+- **get_timeline**: Retrieve your timeline feed
+- **search_posts**: Search for posts by keyword
+- **get_notifications**: Get recent notifications
+- **follow_user**: Follow a user by handle
+- **like_post**: Like a post by URI
+- **repost**: Repost content 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
+pip install -e .
+
+# Run the server
+python -m atproto_mcp
+```
+
+## Usage Example
+
+```python
+from fastmcp import Client
+from atproto_mcp.server import atproto_mcp
+
+async def demo():
+    async with Client(atproto_mcp) as client:
+        # Check status
+        status = await client.call_tool("atproto_status", {})
+        print(f"Connected as: {status['handle']}")
+        
+        # Post to Bluesky
+        post = await client.call_tool("post_to_bluesky", {
+            "text": "Hello from FastMCP! =�"
+        })
+        print(f"Posted: {post['uri']}")
+        
+        # Get timeline
+        timeline = await client.call_tool("get_timeline", {"limit": 5})
+        print(f"Found {timeline['count']} posts")
+        
+        # Search posts
+        results = await client.call_tool("search_posts", {
+            "query": "FastMCP",
+            "limit": 10
+        })
+        print(f"Found {results['count']} posts matching 'FastMCP'")
+```
+
+## Security Note
+
+Store your Bluesky credentials securely in environment variables. Never commit credentials to version control.

examples/atproto_mcp/demo.py

@@ -0,0 +1,101 @@
+#!/usr/bin/env python3
+"""
+Demo script showing ATProto MCP server capabilities.
+"""
+
+import asyncio
+import sys
+from pathlib import Path
+
+# Add the src directory to the path
+sys.path.insert(0, str(Path(__file__).parent / "src"))
+
+from atproto_mcp.server import atproto_mcp
+
+from fastmcp import Client
+
+
+async def main():
+    print("🔵 ATProto MCP Server Demo\n")
+
+    async with Client(atproto_mcp) as client:
+        # 1. Check connection status
+        print("1. Checking connection status...")
+        status = await client.call_tool("atproto_status", {})
+
+        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 (last 3 posts)...")
+        timeline = await client.call_tool("get_timeline", {"limit": 3})
+
+        if timeline.get("success"):
+            print(f"✅ Found {timeline['count']} posts:")
+            for i, post in enumerate(timeline["posts"], 1):
+                print(f"\n   Post {i}:")
+                print(f"   Author: @{post['author']}")
+                print(
+                    f"   Text: {post['text'][:100]}..."
+                    if post["text"] and len(post["text"]) > 100
+                    else f"   Text: {post['text']}"
+                )
+                print(
+                    f"   Likes: {post['likes']} | Reposts: {post['reposts']} | Replies: {post['replies']}"
+                )
+        else:
+            print(f"❌ Failed to get timeline: {timeline.get('error')}")
+
+        # 3. Search for posts
+        print("\n3. Searching for posts about 'Python'...")
+        search = await client.call_tool("search_posts", {"query": "Python", "limit": 3})
+
+        if search.get("success"):
+            print(f"✅ Found {search['count']} posts about Python")
+            if search["posts"]:
+                post = search["posts"][0]
+                print(
+                    f"   Latest by @{post['author']}: {post['text'][:100]}..."
+                    if post["text"] and len(post["text"]) > 100
+                    else f"   Latest by @{post['author']}: {post['text']}"
+                )
+        else:
+            print(f"❌ Search failed: {search.get('error')}")
+
+        # 4. Get notifications
+        print("\n4. Checking notifications...")
+        notifs = await client.call_tool("get_notifications", {"limit": 5})
+
+        if notifs.get("success"):
+            print(f"✅ You have {notifs['count']} recent notifications")
+            unread = sum(1 for n in notifs["notifications"] if not n["is_read"])
+            if unread:
+                print(f"   ({unread} unread)")
+        else:
+            print(f"❌ Failed to get notifications: {notifs.get('error')}")
+
+        # 5. Demo posting (commented out to avoid spam)
+        print("\n5. Posting capability:")
+        print("   To post, you would use:")
+        print(
+            '   await client.call_tool("post_to_bluesky", {"text": "Hello from FastMCP! 🚀"})'
+        )
+
+        # Uncomment to actually post:
+        # post = await client.call_tool("post_to_bluesky", {
+        #     "text": "Testing ATProto MCP server with FastMCP! 🚀"
+        # })
+        # if post.get("success"):
+        #     print(f"✅ Posted successfully: {post['uri']}")
+
+        print("\n✨ Demo complete!")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/atproto_mcp/src/atproto_mcp/server.py

@@ -1,26 +1,210 @@
+from datetime import datetime
+
+from atproto import Client
+
 from atproto_mcp.settings import settings
 from fastmcp import FastMCP
 
 atproto_mcp = FastMCP(
     "ATProto MCP Server",
     dependencies=[
-        "atproto@git+https://github.com/MarshalX/atproto.git@refs/pull/605/head",
-        "pydantic-settings>=2.0.0",
-        "websockets>=15.0.1",
+        "atproto_mcp@git+https://github.com/jlowin/fastmcp.git@atproto-example#subdirectory=examples/atproto_mcp",
     ],
 )
 
+_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
+
+
+@atproto_mcp.tool
+def atproto_status() -> dict:
+    """Check the status of the ATProto connection and current user."""
+    try:
+        client = get_client()
+        profile = client.get_profile(client.me.did)
+        return {
+            "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,
+        }
+    except Exception as e:
+        return {"connected": False, "error": str(e)}
+
+
+@atproto_mcp.tool
+def post_to_bluesky(text: str) -> dict:
+    """Create a new post on Bluesky."""
+    try:
+        client = get_client()
+        post = client.send_post(text=text)
+        return {
+            "success": True,
+            "uri": post.uri,
+            "cid": post.cid,
+            "text": text,
+            "created_at": datetime.now().isoformat(),
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@atproto_mcp.tool
+def get_timeline(limit: int = 10) -> dict:
+    """Get 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(
+                {
+                    "author": post.author.handle,
+                    "text": post.record.text if hasattr(post.record, "text") else None,
+                    "created_at": post.record.created_at
+                    if hasattr(post.record, "created_at")
+                    else None,
+                    "likes": post.like_count,
+                    "reposts": post.repost_count,
+                    "replies": post.reply_count,
+                    "uri": post.uri,
+                }
+            )
+
+        return {
+            "success": True,
+            "count": len(posts),
+            "posts": posts,
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@atproto_mcp.tool
+def search_posts(query: str, limit: int = 10) -> dict:
+    """Search for posts containing specific text."""
+    try:
+        client = get_client()
+        search_results = client.app.bsky.feed.search_posts(
+            q=query,
+            limit=limit,
+        )
+
+        posts = []
+        for post in search_results.posts:
+            posts.append(
+                {
+                    "author": post.author.handle,
+                    "text": post.record.text if hasattr(post.record, "text") else None,
+                    "created_at": post.record.created_at
+                    if hasattr(post.record, "created_at")
+                    else None,
+                    "likes": post.like_count,
+                    "reposts": post.repost_count,
+                    "replies": post.reply_count,
+                    "uri": post.uri,
+                }
+            )
+
+        return {
+            "success": True,
+            "query": query,
+            "count": len(posts),
+            "posts": posts,
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
 
 @atproto_mcp.tool
-def atproto_status() -> str:
-    """Checks the status of the ATProto connection."""
+def get_notifications(limit: int = 10) -> dict:
+    """Get recent notifications for the authenticated user."""
     try:
-        # For now, just verify settings are loaded
-        if settings.atproto_handle and settings.atproto_password:
-            return (
-                f"ATProto credentials configured for handle: {settings.atproto_handle}"
+        client = get_client()
+        notifications = client.app.bsky.notification.list_notifications(limit=limit)
+
+        notifs = []
+        for notif in notifications.notifications:
+            notifs.append(
+                {
+                    "reason": notif.reason,
+                    "author": notif.author.handle if notif.author else None,
+                    "is_read": notif.is_read,
+                    "created_at": notif.indexed_at,
+                    "uri": notif.uri,
+                }
             )
-        else:
-            return "ATProto credentials not configured"
+
+        return {
+            "success": True,
+            "count": len(notifs),
+            "notifications": notifs,
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@atproto_mcp.tool
+def follow_user(handle: str) -> dict:
+    """Follow a user by their handle."""
+    try:
+        client = get_client()
+        # Resolve handle to DID
+        resolved = client.app.bsky.actor.search_actors(q=handle, limit=1)
+        if not resolved.actors:
+            return {"success": False, "error": f"User {handle} not found"}
+
+        user_did = resolved.actors[0].did
+        follow = client.follow(user_did)
+
+        return {
+            "success": True,
+            "followed": handle,
+            "did": user_did,
+            "uri": follow.uri,
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@atproto_mcp.tool
+def like_post(uri: str) -> dict:
+    """Like a post by its AT URI."""
+    try:
+        client = get_client()
+        like = client.like(uri)
+        return {
+            "success": True,
+            "liked_uri": uri,
+            "like_uri": like.uri,
+        }
+    except Exception as e:
+        return {"success": False, "error": str(e)}
+
+
+@atproto_mcp.tool
+def repost(uri: str) -> dict:
+    """Repost a post by its AT URI."""
+    try:
+        client = get_client()
+        repost = client.repost(uri)
+        return {
+            "success": True,
+            "reposted_uri": uri,
+            "repost_uri": repost.uri,
+        }
     except Exception as e:
-        return f"ATProto status check failed: {e}"
+        return {"success": False, "error": str(e)}