ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Refactor ATProto example to use resources and proper type annotations

Browse source 
- 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
This commit is contained in:
zzstoatzz 2025-06-22 18:57:19 -05:00
commit 2eb10094b5
5 changed files with 406 additions and 155 deletions
Download patch file Download diff file Expand all files Collapse all files

49
examples/atproto_mcp/README.md
View file

@ -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.

73
examples/atproto_mcp/demo.py
View file

@ -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!")

254
examples/atproto_mcp/src/atproto_mcp/_atproto.py
View file

@ -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),
)

88
examples/atproto_mcp/src/atproto_mcp/server.py
View file

@ -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)

105
examples/atproto_mcp/src/atproto_mcp/types.py Normal file
View file

@ -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