BlueSky MCP Server
by berlinbra
Verified
- src
- bluesky_mcp
from typing import Any
import asyncio
import json
import os
from atproto import Client
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
API_KEY = os.getenv('BLUESKY_APP_PASSWORD')
IDENTIFIER = os.getenv('BLUESKY_IDENTIFIER')
# if not API_KEY or not IDENTIFIER:
# raise ValueError("BLUESKY_APP_PASSWORD and BLUESKY_IDENTIFIER must be set")
server = Server("bluesky_social")
class BlueSkyClient:
def __init__(self):
self.client = None
async def ensure_client(self):
"""Ensure we have an authenticated client"""
if not self.client:
self.client = Client()
profile = await asyncio.to_thread(
self.client.login,
IDENTIFIER,
API_KEY
)
if not profile:
raise ValueError("Failed to authenticate with BlueSky")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
"""List available tools for BlueSky API integration."""
return [
types.Tool(
name="bluesky_get_profile",
description="Get a user's profile information",
inputSchema={
"type": "object",
"properties": {},
},
),
types.Tool(
name="bluesky_get_posts",
description="Get recent posts from a user",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Maximum number of posts to return (default 50, max 100)",
"default": 50,
},
"cursor": {
"type": "string",
"description": "Pagination cursor for next page of results",
},
},
},
),
types.Tool(
name="bluesky_search_posts",
description="Search for posts on Bluesky",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The search query",
},
"limit": {
"type": "integer",
"description": "Maximum number of posts to return (default 25, max 100)",
"default": 25,
},
"cursor": {
"type": "string",
"description": "Pagination cursor for next page of results",
},
},
"required": ["query"],
},
),
types.Tool(
name="bluesky_get_follows",
description="Get a list of accounts the user follows",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Maximum number of follows to return (default 50, max 100)",
"default": 50,
},
"cursor": {
"type": "string",
"description": "Pagination cursor for next page of results",
},
},
},
),
types.Tool(
name="bluesky_get_followers",
description="Get a list of accounts following the user",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Maximum number of followers to return (default 50, max 100)",
"default": 50,
},
"cursor": {
"type": "string",
"description": "Pagination cursor for next page of results",
},
},
},
),
types.Tool(
name="bluesky_get_liked_posts",
description="Get a list of posts liked by the user",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Maximum number of liked posts to return (default 50, max 100)",
"default": 50,
},
"cursor": {
"type": "string",
"description": "Pagination cursor for next page of results",
},
},
},
),
types.Tool(
name="bluesky_get_personal_feed",
description="Get your personalized Bluesky feed",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Maximum number of feed items to return (default 50, max 100)",
"default": 50,
},
"cursor": {
"type": "string",
"description": "Pagination cursor for next page of results",
},
},
},
),
types.Tool(
name="bluesky_search_profiles",
description="Search for Bluesky profiles",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query string",
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return (default 25, max 100)",
"default": 25,
},
"cursor": {
"type": "string",
"description": "Pagination cursor for next page of results",
},
},
"required": ["query"],
},
),
]
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
"""Handle tool execution requests."""
if not arguments:
arguments = {}
bluesky = BlueSkyClient()
await bluesky.ensure_client()
try:
if name == "bluesky_get_profile":
response = await asyncio.to_thread(
bluesky.client.app.bsky.actor.get_profile,
{'actor': IDENTIFIER}
)
elif name == "bluesky_get_posts":
limit = arguments.get("limit", 50)
cursor = arguments.get("cursor")
response = await asyncio.to_thread(
bluesky.client.app.bsky.feed.get_author_feed,
{'actor': IDENTIFIER, 'limit': limit, 'cursor': cursor}
)
elif name == "bluesky_search_posts":
query = arguments.get("query")
if not query:
return [types.TextContent(type="text", text="Missing required argument: query")]
limit = arguments.get("limit", 25)
cursor = arguments.get("cursor")
response = await asyncio.to_thread(
bluesky.client.app.bsky.feed.search_posts,
{'q': query, 'limit': limit, 'cursor': cursor}
)
elif name == "bluesky_get_follows":
limit = arguments.get("limit", 50)
cursor = arguments.get("cursor")
response = await asyncio.to_thread(
bluesky.client.app.bsky.graph.get_follows,
{'actor': IDENTIFIER, 'limit': limit, 'cursor': cursor}
)
elif name == "bluesky_get_followers":
limit = arguments.get("limit", 50)
cursor = arguments.get("cursor")
response = await asyncio.to_thread(
bluesky.client.app.bsky.graph.get_followers,
{'actor': IDENTIFIER, 'limit': limit, 'cursor': cursor}
)
elif name == "bluesky_get_liked_posts":
limit = arguments.get("limit", 50)
cursor = arguments.get("cursor")
response = await asyncio.to_thread(
bluesky.client.app.bsky.feed.get_likes,
{'uri': IDENTIFIER, 'limit': limit, 'cursor': cursor}
)
elif name == "bluesky_get_personal_feed":
limit = arguments.get("limit", 50)
cursor = arguments.get("cursor")
response = await asyncio.to_thread(
bluesky.client.app.bsky.feed.get_timeline,
{'limit': limit, 'cursor': cursor}
)
elif name == "bluesky_search_profiles":
query = arguments.get("query")
if not query:
return [types.TextContent(type="text", text="Missing required argument: query")]
limit = arguments.get("limit", 25)
cursor = arguments.get("cursor")
response = await asyncio.to_thread(
bluesky.client.app.bsky.actor.search_actors,
{'term': query, 'limit': limit, 'cursor': cursor}
)
else:
return [types.TextContent(type="text", text=f"Unknown tool: {name}")]
return [types.TextContent(type="text", text=json.dumps(response.model_dump(), indent=2))]
except Exception as e:
return [types.TextContent(type="text", text=f"Error: {str(e)}")]
async def main():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="bluesky_social",
server_version="1.0.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
asyncio.run(main())