From 2eb10094b53d8fa6712b2a56f9606eaa77b3b317 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Sun, 22 Jun 2025 18:57:19 -0500 Subject: [PATCH] Refactor ATProto example to use resources and proper type annotations - Add TypedDict definitions for all responses in types.py - Convert read operations to resources (status, timeline, search, notifications) - Keep state-modifying operations as tools (post, follow, like, repost) - Add proper type annotations with Annotated and Field descriptions - Update demo to use read_resource for resources - Better separation of concerns with typed interfaces --- examples/atproto_mcp/README.md | 49 ++-- examples/atproto_mcp/demo.py | 73 +++-- .../atproto_mcp/src/atproto_mcp/_atproto.py | 254 +++++++++++------- .../atproto_mcp/src/atproto_mcp/server.py | 88 ++++-- examples/atproto_mcp/src/atproto_mcp/types.py | 105 ++++++++ 5 files changed, 410 insertions(+), 159 deletions(-) create mode 100644 examples/atproto_mcp/src/atproto_mcp/types.py diff --git a/examples/atproto_mcp/README.md b/examples/atproto_mcp/README.md index 496c40d3d..ecc680181 100644 --- a/examples/atproto_mcp/README.md +++ b/examples/atproto_mcp/README.md @@ -1,16 +1,21 @@ # ATProto MCP Server -This example demonstrates a FastMCP server that provides tools for interacting with the AT Protocol (Bluesky). +This example demonstrates a FastMCP server that provides tools and resources for interacting with the AT Protocol (Bluesky). ## Features -The server provides the following tools: +The server provides two types of capabilities: + +### Resources (Read-only operations) + +- **atproto://profile/status**: Get connection status and profile information +- **atproto://timeline**: Retrieve your timeline feed (last 10 posts) +- **atproto://search/{query}**: Search for posts by keyword +- **atproto://notifications**: Get recent notifications (last 10) + +### Tools (Actions that modify state) -- **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 @@ -43,28 +48,28 @@ 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']}") + # Read resources + status = await client.read_resource("atproto://profile/status") + print(f"Connected as: {status}") - # Post to Bluesky + timeline = await client.read_resource("atproto://timeline?limit=5") + print(f"Timeline: {timeline}") + + # Use tools for actions 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'") + print(f"Posted: {post}") ``` +## Architecture + +The server is organized with: +- `server.py` - Public API with resource and tool definitions +- `_atproto.py` - Private implementation details +- `types.py` - TypedDict definitions for structured responses +- `settings.py` - Configuration management + ## 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 index f4992a696..35b5530fd 100644 --- a/examples/atproto_mcp/demo.py +++ b/examples/atproto_mcp/demo.py @@ -3,8 +3,16 @@ 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 @@ -13,10 +21,12 @@ async def main(enable_posting: bool = False): print("๐Ÿ”ต ATProto MCP Server Demo\n") async with Client(atproto_mcp) as client: - # 1. Check connection status - print("1. Checking connection status...") - result = await client.call_tool("atproto_status", {}) - status = json.loads(result[0].text) if result else {} + # 1. Check connection status (resource) + print("1. Checking connection status (resource)...") + 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']}") @@ -27,10 +37,12 @@ async def main(enable_posting: bool = False): print(f"โŒ Connection failed: {status.get('error')}") return - # 2. Get timeline - print("\n2. Getting timeline (last 3 posts)...") - result = await client.call_tool("get_timeline", {"limit": 3}) - timeline = json.loads(result[0].text) if result else {} + # 2. Get timeline (resource with parameter) + print("\n2. Getting timeline (resource)...") + result = await client.read_resource("atproto://timeline") + timeline: TimelineResult = ( + json.loads(result[0].text) if result else cast(TimelineResult, {}) + ) if timeline.get("success"): print(f"โœ… Found {timeline['count']} posts:") @@ -48,10 +60,12 @@ async def main(enable_posting: bool = False): else: print(f"โŒ Failed to get timeline: {timeline.get('error')}") - # 3. Search for posts - print("\n3. Searching for posts about 'Python'...") - result = await client.call_tool("search_posts", {"query": "Python", "limit": 3}) - search = json.loads(result[0].text) if result else {} + # 3. Search for posts (resource with template) + print("\n3. Searching for posts about 'Python' (template resource)...") + result = await client.read_resource("atproto://search/Python") + search: SearchResult = ( + json.loads(result[0].text) if result else cast(SearchResult, {}) + ) if search.get("success"): print(f"โœ… Found {search['count']} posts about Python") @@ -65,10 +79,12 @@ async def main(enable_posting: bool = False): else: print(f"โŒ Search failed: {search.get('error')}") - # 4. Get notifications - print("\n4. Checking notifications...") - result = await client.call_tool("get_notifications", {"limit": 5}) - notifs = json.loads(result[0].text) if result else {} + # 4. Get notifications (resource) + print("\n4. Checking notifications (resource)...") + 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']} recent notifications") @@ -78,16 +94,18 @@ async def main(enable_posting: bool = False): else: print(f"โŒ Failed to get notifications: {notifs.get('error')}") - # 5. Demo posting + # 5. Demo posting (tool) if enable_posting: - print("\n5. Creating a test post...") - post = await client.call_tool( + print("\n5. Creating a test post (tool)...") + post_result = await client.call_tool( "post_to_bluesky", { "text": "๐Ÿงช Testing the ATProto MCP server demo! This post was created programmatically using FastMCP. #FastMCP #ATProto" }, ) - result = json.loads(post[0].text) if post else {} + result: PostResult = ( + json.loads(post_result[0].text) if post_result else cast(PostResult, {}) + ) if result.get("success"): print("โœ… Posted successfully!") print(f" URI: {result['uri']}") @@ -95,10 +113,23 @@ async def main(enable_posting: bool = False): else: print(f"โŒ Failed to post: {result.get('error')}") else: - print("\n5. Posting capability:") + print("\n5. Posting capability (tool):") print(" To enable posting, run with --post flag") print(" Example: python demo.py --post") + # 6. Show available resources and tools + print("\n6. Available capabilities:") + print(" Resources (read-only):") + print(" - atproto://profile/status - Profile information") + print(" - atproto://timeline - Timeline feed") + print(" - atproto://search/{query} - Search posts") + print(" - atproto://notifications - Recent notifications") + print(" Tools (actions):") + print(" - post_to_bluesky - Create a new post") + print(" - follow_user - Follow a user") + print(" - like_post - Like a post") + print(" - repost - Repost content") + print("\nโœจ Demo complete!") diff --git a/examples/atproto_mcp/src/atproto_mcp/_atproto.py b/examples/atproto_mcp/src/atproto_mcp/_atproto.py index 3b052ca67..c950fe619 100644 --- a/examples/atproto_mcp/src/atproto_mcp/_atproto.py +++ b/examples/atproto_mcp/src/atproto_mcp/_atproto.py @@ -5,6 +5,18 @@ from datetime import datetime from atproto import Client from atproto_mcp.settings import settings +from atproto_mcp.types import ( + FollowResult, + LikeResult, + Notification, + NotificationsResult, + Post, + PostResult, + ProfileInfo, + RepostResult, + SearchResult, + TimelineResult, +) _client: Client | None = None @@ -18,41 +30,59 @@ def get_client() -> Client: return _client -def get_profile_info() -> dict: +def get_profile_info() -> ProfileInfo: """Get profile information for the authenticated 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, - } + 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 {"connected": False, "error": str(e)} + return ProfileInfo( + connected=False, + handle=None, + display_name=None, + did=None, + followers=None, + following=None, + posts=None, + error=str(e), + ) -def create_post(text: str) -> dict: +def create_post(text: str) -> PostResult: """Create a new post.""" 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(), - } + return PostResult( + success=True, + uri=post.uri, + cid=post.cid, + text=text, + created_at=datetime.now().isoformat(), + error=None, + ) except Exception as e: - return {"success": False, "error": str(e)} + return PostResult( + success=False, + uri=None, + cid=None, + text=None, + created_at=None, + error=str(e), + ) -def fetch_timeline(limit: int = 10) -> dict: +def fetch_timeline(limit: int = 10) -> TimelineResult: """Fetch the authenticated user's timeline.""" try: client = get_client() @@ -62,29 +92,35 @@ def fetch_timeline(limit: int = 10) -> dict: 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 + Post( + 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, - } + likes=post.like_count, + reposts=post.repost_count, + replies=post.reply_count, + uri=post.uri, + ) ) - return { - "success": True, - "count": len(posts), - "posts": posts, - } + return TimelineResult( + success=True, + count=len(posts), + posts=posts, + error=None, + ) except Exception as e: - return {"success": False, "error": str(e)} + return TimelineResult( + success=False, + count=0, + posts=[], + error=str(e), + ) -def search_for_posts(query: str, limit: int = 10) -> dict: +def search_for_posts(query: str, limit: int = 10) -> SearchResult: """Search for posts containing specific text.""" try: client = get_client() @@ -95,30 +131,37 @@ def search_for_posts(query: str, limit: int = 10) -> dict: 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 + Post( + 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, - } + 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, - } + return SearchResult( + success=True, + query=query, + count=len(posts), + posts=posts, + error=None, + ) except Exception as e: - return {"success": False, "error": str(e)} + return SearchResult( + success=False, + query=query, + count=0, + posts=[], + error=str(e), + ) -def fetch_notifications(limit: int = 10) -> dict: +def fetch_notifications(limit: int = 10) -> NotificationsResult: """Get recent notifications.""" try: client = get_client() @@ -129,69 +172,100 @@ def fetch_notifications(limit: int = 10) -> dict: 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, - } + Notification( + 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, + ) ) - return { - "success": True, - "count": len(notifs), - "notifications": notifs, - } + return NotificationsResult( + success=True, + count=len(notifs), + notifications=notifs, + error=None, + ) except Exception as e: - return {"success": False, "error": str(e)} + return NotificationsResult( + success=False, + count=0, + notifications=[], + error=str(e), + ) -def follow_user_by_handle(handle: str) -> dict: +def follow_user_by_handle(handle: str) -> FollowResult: """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"} + return FollowResult( + success=False, + followed=None, + did=None, + uri=None, + 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, - } + return FollowResult( + success=True, + followed=handle, + did=user_did, + uri=follow.uri, + error=None, + ) except Exception as e: - return {"success": False, "error": str(e)} + return FollowResult( + success=False, + followed=None, + did=None, + uri=None, + error=str(e), + ) -def like_post_by_uri(uri: str) -> dict: +def like_post_by_uri(uri: str) -> LikeResult: """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, - } + return LikeResult( + success=True, + liked_uri=uri, + like_uri=like.uri, + error=None, + ) except Exception as e: - return {"success": False, "error": str(e)} + return LikeResult( + success=False, + liked_uri=None, + like_uri=None, + error=str(e), + ) -def repost_by_uri(uri: str) -> dict: +def repost_by_uri(uri: str) -> RepostResult: """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, - } + return RepostResult( + success=True, + reposted_uri=uri, + repost_uri=repost.uri, + error=None, + ) except Exception as e: - return {"success": False, "error": str(e)} + return RepostResult( + success=False, + reposted_uri=None, + repost_uri=None, + error=str(e), + ) diff --git a/examples/atproto_mcp/src/atproto_mcp/server.py b/examples/atproto_mcp/src/atproto_mcp/server.py index e1067c5c5..73687f27f 100644 --- a/examples/atproto_mcp/src/atproto_mcp/server.py +++ b/examples/atproto_mcp/src/atproto_mcp/server.py @@ -1,6 +1,20 @@ -"""ATProto MCP Server - Public API exposing Bluesky tools.""" +"""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.types import ( + FollowResult, + LikeResult, + NotificationsResult, + PostResult, + ProfileInfo, + RepostResult, + SearchResult, + TimelineResult, +) from fastmcp import FastMCP atproto_mcp = FastMCP( @@ -11,49 +25,71 @@ atproto_mcp = FastMCP( ) -@atproto_mcp.tool -def atproto_status() -> dict: - """Check the status of the ATProto connection and current user.""" +# 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(10) + + +@atproto_mcp.resource("atproto://search/{query}") +def search_posts( + query: Annotated[str, Field(description="Search query for posts")], + limit: Annotated[ + int, Field(default=10, ge=1, le=100, description="Number of results to return") + ] = 10, +) -> SearchResult: + """Search for posts containing specific text.""" + return _atproto.search_for_posts(query, limit) + + +@atproto_mcp.resource("atproto://notifications") +def get_notifications() -> NotificationsResult: + """Get recent notifications for the authenticated user.""" + return _atproto.fetch_notifications(10) + + +# Tools - actions that modify state @atproto_mcp.tool -def post_to_bluesky(text: str) -> dict: +def post_to_bluesky( + text: Annotated[ + str, Field(max_length=300, description="The text content of the post") + ], +) -> PostResult: """Create a new post on Bluesky.""" return _atproto.create_post(text) @atproto_mcp.tool -def get_timeline(limit: int = 10) -> dict: - """Get the authenticated user's timeline.""" - return _atproto.fetch_timeline(limit) - - -@atproto_mcp.tool -def search_posts(query: str, limit: int = 10) -> dict: - """Search for posts containing specific text.""" - return _atproto.search_for_posts(query, limit) - - -@atproto_mcp.tool -def get_notifications(limit: int = 10) -> dict: - """Get recent notifications for the authenticated user.""" - return _atproto.fetch_notifications(limit) - - -@atproto_mcp.tool -def follow_user(handle: str) -> dict: +def follow_user( + 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_post(uri: str) -> dict: +def like_post( + 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: str) -> dict: +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) 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..4cf726448 --- /dev/null +++ b/examples/atproto_mcp/src/atproto_mcp/types.py @@ -0,0 +1,105 @@ +"""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 + + +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 + created_at: str + uri: 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 + followed: 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