diff --git a/examples/atproto_mcp/README.md b/examples/atproto_mcp/README.md new file mode 100644 index 000000000..2ca9a4b72 --- /dev/null +++ b/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. \ No newline at end of file diff --git a/examples/atproto_mcp/demo.py b/examples/atproto_mcp/demo.py new file mode 100644 index 000000000..0184c27f9 --- /dev/null +++ b/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)) diff --git a/examples/atproto_mcp/pyproject.toml b/examples/atproto_mcp/pyproject.toml new file mode 100644 index 000000000..42682ef18 --- /dev/null +++ b/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 } diff --git a/examples/atproto_mcp/src/atproto_mcp/__init__.py b/examples/atproto_mcp/src/atproto_mcp/__init__.py new file mode 100644 index 000000000..9752f9b8c --- /dev/null +++ b/examples/atproto_mcp/src/atproto_mcp/__init__.py @@ -0,0 +1,3 @@ +from atproto_mcp.settings import settings + +__all__ = ["settings"] diff --git a/examples/atproto_mcp/src/atproto_mcp/__main__.py b/examples/atproto_mcp/src/atproto_mcp/__main__.py new file mode 100644 index 000000000..bb4c12e7a --- /dev/null +++ b/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() diff --git a/examples/atproto_mcp/src/atproto_mcp/_atproto/__init__.py b/examples/atproto_mcp/src/atproto_mcp/_atproto/__init__.py new file mode 100644 index 000000000..6418f7d43 --- /dev/null +++ b/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", +] diff --git a/examples/atproto_mcp/src/atproto_mcp/_atproto/_client.py b/examples/atproto_mcp/src/atproto_mcp/_atproto/_client.py new file mode 100644 index 000000000..40ee8e160 --- /dev/null +++ b/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 diff --git a/examples/atproto_mcp/src/atproto_mcp/_atproto/_posts.py b/examples/atproto_mcp/src/atproto_mcp/_atproto/_posts.py new file mode 100644 index 000000000..7b53e51fe --- /dev/null +++ b/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, + ) diff --git a/examples/atproto_mcp/src/atproto_mcp/_atproto/_profile.py b/examples/atproto_mcp/src/atproto_mcp/_atproto/_profile.py new file mode 100644 index 000000000..956ae5412 --- /dev/null +++ b/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), + ) diff --git a/examples/atproto_mcp/src/atproto_mcp/_atproto/_read.py b/examples/atproto_mcp/src/atproto_mcp/_atproto/_read.py new file mode 100644 index 000000000..189185a4a --- /dev/null +++ b/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), + ) diff --git a/examples/atproto_mcp/src/atproto_mcp/_atproto/_social.py b/examples/atproto_mcp/src/atproto_mcp/_atproto/_social.py new file mode 100644 index 000000000..87bd02976 --- /dev/null +++ b/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), + ) diff --git a/examples/atproto_mcp/src/atproto_mcp/py.typed b/examples/atproto_mcp/src/atproto_mcp/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/examples/atproto_mcp/src/atproto_mcp/server.py b/examples/atproto_mcp/src/atproto_mcp/server.py new file mode 100644 index 000000000..a0380d4f6 --- /dev/null +++ b/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) diff --git a/examples/atproto_mcp/src/atproto_mcp/settings.py b/examples/atproto_mcp/src/atproto_mcp/settings.py new file mode 100644 index 000000000..9eed40837 --- /dev/null +++ b/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() diff --git a/examples/atproto_mcp/src/atproto_mcp/types.py b/examples/atproto_mcp/src/atproto_mcp/types.py new file mode 100644 index 000000000..1405bda41 --- /dev/null +++ b/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