# Directory Structure
```
├── .gitignore
├── .python-version
├── LICENSE
├── pyproject.toml
├── README.md
├── src
│ └── bluesky_mcp
│ ├── __init__.py
│ └── server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# BlueSky MCP Server
A Model Context Protocol (MCP) server that provides access to [BlueSky](https://bsky.app) social network data through its official API. This server implements a standardized interface for retrieving user profiles and social graph information.
<a href="https://glama.ai/mcp/servers/bxvvsqt34k"><img width="380" height="200" src="https://glama.ai/mcp/servers/bxvvsqt34k/badge" alt="BlueSky Server MCP server" /></a>
## Features
- Fetch detailed user profile information
- Retrieve user following lists with pagination
- Built-in authentication handling and session management
- Comprehensive error handling
## Installation
#### Claude Desktop
- On MacOS: `~/Library/Application\ Support/Claude/claude_desktop_config.json`
- On Windows: `%APPDATA%/Claude/claude_desktop_config.json`
<details>
<summary>Development/Unpublished Servers Configuration</summary>
```json
"mcpServers": {
"bluesky-mcp": {
"command": "uv",
"args": [
"--directory",
"C:\\Users\\{INSERT_USER}\\YOUR\\PATH\\TO\\bluesky-mcp\\bluesky-mcp",
"run",
"src/bluesky_mcp/server.py"
],
"env": {
"BLUESKY_IDENTIFIER": "your.handle.bsky.social",
"BLUESKY_APP_PASSWORD": "your-app-password"
}
}
}
```
</details>
### Running Locally
#### Install Libraries
```
uv pip install -e .
```
### Running
After connecting Claude client with the MCP tool via json file and installing the packages, Claude should see the server's mcp tools:
You can run the sever yourself via:
In bluesky_mcp repo:
```
uv run src/bluesky_mcp/server.py
```
*if you want to run the server inspector along with the server:
```
npx @modelcontextprotocol/inspector uv --directory C:\\Users\\{INSERT_USER}\\YOUR\\PATH\\TO\\bluesky-mcp run src/bluesky_mcp/server.py
```
## Available Tools
The server implements two tools:
- `get-profile`: Get detailed profile information for a BlueSky user
- `get-follows`: Get a list of accounts that a specified user follows
### get-profile
Retrieves detailed profile information for a given BlueSky user.
**Input Schema:**
```json
{
"handle": {
"type": "string",
"description": "The user's handle (e.g., 'alice.bsky.social')"
}
}
```
**Example Response:**
```
Profile information for alice.bsky.social:
Handle: alice.bsky.social
Display Name: Alice
Description: Just a BlueSky user sharing thoughts
Followers: 1234
Following: 567
Posts: 789
```
### get-follows
Retrieves a list of accounts that a specified user follows, with support for pagination.
**Input Schema:**
```json
{
"actor": {
"type": "string",
"description": "The user's handle (e.g., 'alice.bsky.social')"
},
"limit": {
"type": "integer",
"description": "Maximum number of results to return",
"default": 50,
"minimum": 1,
"maximum": 100
},
"cursor": {
"type": "string",
"description": "Pagination cursor",
"optional": true
}
}
```
**Example Response:**
```
Follows for alice.bsky.social:
Follows:
Handle: bob.bsky.social
Display Name: Bob
---
Handle: carol.bsky.social
Display Name: Carol
---
Handle: dave.bsky.social
Display Name: Dave
---
More results available. Use cursor: bafygeia...
```
## Error Handling
The server includes comprehensive error handling for various scenarios:
- Authentication failures
- Rate limiting
- Network connectivity issues
- Invalid parameters
- Timeout handling
- Malformed responses
Error messages are returned in a clear, human-readable format.
## Prerequisites
- Python 3.12 or higher
- httpx
- mcp
## Authentication
To use this MCP server, you need to:
1. Create a BlueSky account if you don't have one
2. Generate an App Password in your BlueSky account settings
3. Set the following environment variables:
- `BLUESKY_IDENTIFIER`: Your BlueSky handle (e.g., "username.bsky.social")
- `BLUESKY_APP_PASSWORD`: Your generated App Password
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
## License
This MCP server is licensed under the MIT License.
This means you are free to use, modify, and distribute the software, subject to the terms and conditions of the MIT License. For more details, please see the LICENSE file in the project repository.
```
--------------------------------------------------------------------------------
/src/bluesky_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
from .server import main
__all__ = ["main"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "bluesky_mcp"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
"mcp>=0.1.0",
"atproto>=0.0.47",
"pydantic>=2.0.0"
]
[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"
[project.scripts]
bluesky-mcp = "bluesky_mcp:main"
[[project.authors]]
name = "berlinbra"
email = "[email protected]"
```
--------------------------------------------------------------------------------
/src/bluesky_mcp/server.py:
--------------------------------------------------------------------------------
```python
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())
```