"""
SwipeBuilder MCP Server
Standalone MCP server that provides access to SwipeBuilder's ad swipe library
via the public REST API. This server runs independently and communicates with
the SwipeBuilder API using httpx.
Usage:
# With Claude Desktop - add to config:
{
"mcpServers": {
"swipebuilder": {
"command": "swipebuilder-mcp",
"env": {
"SWIPEBUILDER_API_KEY": "sb_live_xxx"
}
}
}
}
# Or run directly:
SWIPEBUILDER_API_KEY=sb_live_xxx swipebuilder-mcp
"""
import os
import logging
from typing import Optional
import httpx
from mcp.server.fastmcp import FastMCP
logger = logging.getLogger(__name__)
# Configuration
API_BASE_URL = os.getenv("SWIPEBUILDER_API_URL", "https://api.swipebuilder.io/v1")
API_KEY = os.getenv("SWIPEBUILDER_API_KEY", "")
REQUEST_TIMEOUT = 30.0
# Initialize MCP server
mcp = FastMCP(
"SwipeBuilder",
instructions="""
SwipeBuilder MCP Server - Access your ad swipe library programmatically.
Available tools:
- get_swipes: Retrieve ads from your swipe file with optional filters
- get_swipe: Get a single ad by ID
- get_collections: List your collections/folders
- get_collection: Get collection details by ID
- get_collection_items: Get ads within a specific collection
- get_usage: Check your API credit usage
All tools require a valid API key. Set via SWIPEBUILDER_API_KEY environment variable,
or pass directly to each tool call.
Credits are consumed for each ad returned (1 credit per ad).
Collections themselves don't consume credits.
"""
)
def get_api_key(provided_key: Optional[str] = None) -> str:
"""Get API key from parameter or environment."""
key = provided_key or API_KEY
if not key:
raise ValueError(
"No API key provided. Set SWIPEBUILDER_API_KEY environment variable "
"or pass api_key parameter to the tool."
)
return key
async def api_request(
method: str,
endpoint: str,
api_key: str,
params: Optional[dict] = None
) -> dict:
"""Make an authenticated request to the SwipeBuilder API."""
url = f"{API_BASE_URL}{endpoint}"
headers = {"X-API-Key": api_key}
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
if method.upper() == "GET":
response = await client.get(url, headers=headers, params=params)
else:
response = await client.request(method, url, headers=headers, params=params)
if response.status_code == 401:
return {"error": "Invalid or inactive API key"}
elif response.status_code == 403:
return {"error": "Insufficient credits or permissions"}
elif response.status_code == 404:
return {"error": "Resource not found"}
elif response.status_code == 429:
return {"error": "Rate limit exceeded. Please slow down requests."}
elif response.status_code >= 400:
return {"error": f"API error: {response.status_code} - {response.text}"}
return response.json()
@mcp.tool()
async def get_swipes(
api_key: Optional[str] = None,
platform: Optional[str] = None,
format: Optional[str] = None,
limit: int = 50,
cursor: Optional[str] = None
) -> dict:
"""
Get swipes from your swipe file.
Args:
api_key: Your SwipeBuilder API key (sb_live_xxx or sb_test_xxx).
Optional if SWIPEBUILDER_API_KEY env var is set.
platform: Filter by platform (facebook, tiktok, instagram, linkedin, google)
format: Filter by format (image, video, carousel)
limit: Number of results (1-100, default 50)
cursor: Pagination cursor from previous response
Returns:
Dictionary with swipes data, pagination info, and credit usage
Credit Usage: 1 credit per swipe returned
"""
try:
key = get_api_key(api_key)
except ValueError as e:
return {"error": str(e)}
params: dict[str, str | int] = {"limit": min(limit, 100)}
if platform:
params["platform"] = platform
if format:
params["format"] = format
if cursor:
params["cursor"] = cursor
result = await api_request("GET", "/swipes", key, params)
if "error" not in result:
# Add credits_used info based on returned count
data = result.get("data", [])
result["credits_used"] = len(data)
return result
@mcp.tool()
async def get_swipe(
swipe_id: str,
api_key: Optional[str] = None
) -> dict:
"""
Get a single swipe by ID.
Args:
swipe_id: The ID of the swipe to retrieve
api_key: Your SwipeBuilder API key. Optional if SWIPEBUILDER_API_KEY env var is set.
Returns:
Dictionary with swipe data or error message
Credit Usage: 1 credit
"""
try:
key = get_api_key(api_key)
except ValueError as e:
return {"error": str(e)}
result = await api_request("GET", f"/swipes/{swipe_id}", key)
if "error" not in result:
result["credits_used"] = 1
return result
@mcp.tool()
async def get_collections(
api_key: Optional[str] = None,
parent_id: Optional[str] = None,
limit: int = 50,
cursor: Optional[str] = None
) -> dict:
"""
Get your collections (folders).
Args:
api_key: Your SwipeBuilder API key. Optional if SWIPEBUILDER_API_KEY env var is set.
parent_id: Filter by parent ('root' for root-level, collection ID for children, None for all)
limit: Number of results (1-100, default 50)
cursor: Pagination cursor from previous response
Returns:
Dictionary with collections data and pagination info
Credit Usage: No credits consumed
"""
try:
key = get_api_key(api_key)
except ValueError as e:
return {"error": str(e)}
params: dict[str, str | int] = {"limit": min(limit, 100)}
if parent_id:
params["parent_id"] = parent_id
if cursor:
params["cursor"] = cursor
return await api_request("GET", "/collections", key, params)
@mcp.tool()
async def get_collection(
collection_id: str,
api_key: Optional[str] = None
) -> dict:
"""
Get a single collection by ID.
Args:
collection_id: The ID of the collection to retrieve
api_key: Your SwipeBuilder API key. Optional if SWIPEBUILDER_API_KEY env var is set.
Returns:
Dictionary with collection data or error message
Credit Usage: No credits consumed
"""
try:
key = get_api_key(api_key)
except ValueError as e:
return {"error": str(e)}
return await api_request("GET", f"/collections/{collection_id}", key)
@mcp.tool()
async def get_collection_items(
collection_id: str,
api_key: Optional[str] = None,
limit: int = 50,
cursor: Optional[str] = None
) -> dict:
"""
Get items (swipes) in a collection.
Args:
collection_id: The ID of the collection
api_key: Your SwipeBuilder API key. Optional if SWIPEBUILDER_API_KEY env var is set.
limit: Number of results (1-100, default 50)
cursor: Pagination cursor from previous response
Returns:
Dictionary with collection info, items data, and pagination
Credit Usage: 1 credit per swipe returned
"""
try:
key = get_api_key(api_key)
except ValueError as e:
return {"error": str(e)}
params: dict[str, str | int] = {"limit": min(limit, 100)}
if cursor:
params["cursor"] = cursor
result = await api_request("GET", f"/collections/{collection_id}/items", key, params)
if "error" not in result:
# Add credits_used info based on returned count
data = result.get("data", [])
result["credits_used"] = len(data)
return result
@mcp.tool()
async def get_usage(api_key: Optional[str] = None) -> dict:
"""
Get your API credit usage for the current month.
Args:
api_key: Your SwipeBuilder API key. Optional if SWIPEBUILDER_API_KEY env var is set.
Returns:
Dictionary with usage statistics including:
- credits_used: Credits consumed this month
- credits_limit: Monthly credit limit
- credits_remaining: Credits still available
- period_start: Start of billing period
- period_end: End of billing period
Credit Usage: No credits consumed
"""
try:
key = get_api_key(api_key)
except ValueError as e:
return {"error": str(e)}
return await api_request("GET", "/usage", key)
def main():
"""Run the MCP server via stdio transport."""
mcp.run()
if __name__ == "__main__":
main()
