mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-28 02:10:38 +02:00
Refactor ATProto example with modular design and improved demo
- Create _atproto.py module for private implementation details - Clean server.py to only expose public API (tools) - Update demo with --post flag instead of comments - Better separation of concerns
This commit is contained in:
zzstoatzz
parent
4421bafb2a
commit
5c23ac8ee9
3 changed files with 238 additions and 177 deletions
|
|
@ -1,5 +1,6 @@
|
|||
"""Demo script showing ATProto MCP server capabilities."""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
|
|
@ -8,7 +9,7 @@ from atproto_mcp.server import atproto_mcp
|
|||
from fastmcp import Client
|
||||
|
||||
|
||||
async def main():
|
||||
async def main(enable_posting: bool = False):
|
||||
print("🔵 ATProto MCP Server Demo\n")
|
||||
|
||||
async with Client(atproto_mcp) as client:
|
||||
|
|
@ -77,22 +78,37 @@ async def main():
|
|||
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']}")
|
||||
# 5. Demo posting
|
||||
if enable_posting:
|
||||
print("\n5. Creating a test post...")
|
||||
post = 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 {}
|
||||
if result.get("success"):
|
||||
print("✅ Posted successfully!")
|
||||
print(f" URI: {result['uri']}")
|
||||
print(f" Created at: {result['created_at']}")
|
||||
else:
|
||||
print(f"❌ Failed to post: {result.get('error')}")
|
||||
else:
|
||||
print("\n5. Posting capability:")
|
||||
print(" To enable posting, run with --post flag")
|
||||
print(" Example: python demo.py --post")
|
||||
|
||||
print("\n✨ Demo complete!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
parser = argparse.ArgumentParser(description="ATProto MCP Server Demo")
|
||||
parser.add_argument(
|
||||
"--post",
|
||||
action="store_true",
|
||||
help="Enable posting a test message to Bluesky",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
asyncio.run(main(enable_posting=args.post))
|
||||
|
|
|
|||
197
examples/atproto_mcp/src/atproto_mcp/_atproto.py
Normal file
197
examples/atproto_mcp/src/atproto_mcp/_atproto.py
Normal file
|
|
@ -0,0 +1,197 @@
|
|||
"""Private ATProto implementation details."""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
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
|
||||
|
||||
|
||||
def get_profile_info() -> dict:
|
||||
"""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,
|
||||
}
|
||||
except Exception as e:
|
||||
return {"connected": False, "error": str(e)}
|
||||
|
||||
|
||||
def create_post(text: str) -> dict:
|
||||
"""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(),
|
||||
}
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
|
||||
def fetch_timeline(limit: int = 10) -> dict:
|
||||
"""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(
|
||||
{
|
||||
"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)}
|
||||
|
||||
|
||||
def search_for_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(
|
||||
params={"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)}
|
||||
|
||||
|
||||
def fetch_notifications(limit: int = 10) -> dict:
|
||||
"""Get recent notifications."""
|
||||
try:
|
||||
client = get_client()
|
||||
notifications = client.app.bsky.notification.list_notifications(
|
||||
params={"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,
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"count": len(notifs),
|
||||
"notifications": notifs,
|
||||
}
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
|
||||
def follow_user_by_handle(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)}
|
||||
|
||||
|
||||
def like_post_by_uri(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)}
|
||||
|
||||
|
||||
def repost_by_uri(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 {"success": False, "error": str(e)}
|
||||
|
|
@ -1,8 +1,6 @@
|
|||
from datetime import datetime
|
||||
"""ATProto MCP Server - Public API exposing Bluesky tools."""
|
||||
|
||||
from atproto import Client
|
||||
|
||||
from atproto_mcp.settings import settings
|
||||
from atproto_mcp import _atproto
|
||||
from fastmcp import FastMCP
|
||||
|
||||
atproto_mcp = FastMCP(
|
||||
|
|
@ -12,200 +10,50 @@ atproto_mcp = FastMCP(
|
|||
],
|
||||
)
|
||||
|
||||
_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)}
|
||||
return _atproto.get_profile_info()
|
||||
|
||||
|
||||
@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)}
|
||||
return _atproto.create_post(text)
|
||||
|
||||
|
||||
@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)}
|
||||
return _atproto.fetch_timeline(limit)
|
||||
|
||||
|
||||
@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(
|
||||
params={"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)}
|
||||
return _atproto.search_for_posts(query, limit)
|
||||
|
||||
|
||||
@atproto_mcp.tool
|
||||
def get_notifications(limit: int = 10) -> dict:
|
||||
"""Get recent notifications for the authenticated user."""
|
||||
try:
|
||||
client = get_client()
|
||||
notifications = client.app.bsky.notification.list_notifications(
|
||||
params={"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,
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"count": len(notifs),
|
||||
"notifications": notifs,
|
||||
}
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e)}
|
||||
return _atproto.fetch_notifications(limit)
|
||||
|
||||
|
||||
@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)}
|
||||
return _atproto.follow_user_by_handle(handle)
|
||||
|
||||
|
||||
@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)}
|
||||
return _atproto.like_post_by_uri(uri)
|
||||
|
||||
|
||||
@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 {"success": False, "error": str(e)}
|
||||
return _atproto.repost_by_uri(uri)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue