mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-20 20:44:17 +02:00
Implement full ATProto functionality with 8 tools for Bluesky interaction
This commit is contained in:
zzstoatzz
parent
8faf3f5721
commit
40226a8fb7
3 changed files with 367 additions and 12 deletions
|
|
@ -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.
|
||||
101
examples/atproto_mcp/demo.py
Normal file
101
examples/atproto_mcp/demo.py
Normal file
|
|
@ -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())
|
||||
|
|
@ -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() -> str:
|
||||
"""Checks the status of the ATProto connection."""
|
||||
def atproto_status() -> dict:
|
||||
"""Check the status of the ATProto connection and current 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}"
|
||||
)
|
||||
else:
|
||||
return "ATProto credentials not configured"
|
||||
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 f"ATProto status check failed: {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 get_notifications(limit: int = 10) -> dict:
|
||||
"""Get recent notifications for the authenticated user."""
|
||||
try:
|
||||
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,
|
||||
}
|
||||
)
|
||||
|
||||
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 {"success": False, "error": str(e)}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue