This is page 1 of 5. Use http://codebase.md/nictuku/meta-ads-mcp?page={x} to view the full context.
# Directory Structure
```
├── .github
│ └── workflows
│ ├── publish-mcp.yml
│ ├── publish.yml
│ └── test.yml
├── .gitignore
├── .python-version
├── .uv.toml
├── CUSTOM_META_APP.md
├── Dockerfile
├── examples
│ ├── example_http_client.py
│ └── README.md
├── future_improvements.md
├── images
│ └── meta-ads-example.png
├── LICENSE
├── LOCAL_INSTALLATION.md
├── meta_ads_auth.sh
├── meta_ads_mcp
│ ├── __init__.py
│ ├── __main__.py
│ └── core
│ ├── __init__.py
│ ├── accounts.py
│ ├── ads_library.py
│ ├── ads.py
│ ├── adsets.py
│ ├── api.py
│ ├── auth.py
│ ├── authentication.py
│ ├── budget_schedules.py
│ ├── callback_server.py
│ ├── campaigns.py
│ ├── duplication.py
│ ├── http_auth_integration.py
│ ├── insights.py
│ ├── openai_deep_research.py
│ ├── pipeboard_auth.py
│ ├── reports.py
│ ├── resources.py
│ ├── server.py
│ ├── targeting.py
│ └── utils.py
├── META_API_NOTES.md
├── poetry.lock
├── pyproject.toml
├── README.md
├── RELEASE.md
├── requirements.txt
├── server.json
├── setup.py
├── smithery.yaml
├── STREAMABLE_HTTP_SETUP.md
└── tests
├── __init__.py
├── conftest.py
├── e2e_account_info_search_issue.py
├── README_REGRESSION_TESTS.md
├── README.md
├── test_account_info_access_fix.py
├── test_account_search.py
├── test_budget_update_e2e.py
├── test_budget_update.py
├── test_create_ad_creative_simple.py
├── test_create_simple_creative_e2e.py
├── test_dsa_beneficiary.py
├── test_dsa_integration.py
├── test_duplication_regression.py
├── test_duplication.py
├── test_dynamic_creatives.py
├── test_estimate_audience_size_e2e.py
├── test_estimate_audience_size.py
├── test_get_account_pages.py
├── test_get_ad_creatives_fix.py
├── test_get_ad_image_quality_improvements.py
├── test_get_ad_image_regression.py
├── test_http_transport.py
├── test_insights_actions_and_values_e2e.py
├── test_insights_pagination.py
├── test_integration_openai_mcp.py
├── test_is_dynamic_creative_adset.py
├── test_mobile_app_adset_creation.py
├── test_mobile_app_adset_issue.py
├── test_openai_mcp_deep_research.py
├── test_openai.py
├── test_page_discovery_integration.py
├── test_page_discovery.py
├── test_targeting_search_e2e.py
├── test_targeting.py
├── test_update_ad_creative_id.py
└── test_upload_ad_image.py
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.10
```
--------------------------------------------------------------------------------
/.uv.toml:
--------------------------------------------------------------------------------
```toml
[uv]
python-path = ["python3.11"]
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
.cursor/rules/meta-ads-credentials.mdc
.DS_Store
# Development environment files
.python-version
poetry.lock
uv.lock
.uv.toml
.cursor/
# Generated content
ad_creatives/
# Debug and development files
debug/
*.pyc
.pytest_cache/
# Keep organized directories but exclude some debug content
!debug/README.md
internal/
uv.lock
```
--------------------------------------------------------------------------------
/examples/README.md:
--------------------------------------------------------------------------------
```markdown
# Meta Ads MCP Examples
This directory contains example scripts and usage demonstrations for the Meta Ads MCP server.
## Files
### `http_client.py`
A complete example HTTP client that demonstrates how to interact with the Meta Ads MCP server using the HTTP transport.
**Features:**
- Shows how to authenticate with Pipeboard tokens or Meta access tokens
- Demonstrates all basic MCP operations (initialize, list tools, call tools)
- Includes error handling and response formatting
- Ready-to-use client class for integration
**Usage:**
```bash
# Start the MCP server
python -m meta_ads_mcp --transport streamable-http --port 8080
# Run the example (in another terminal)
cd examples
python http_client.py
```
**Authentication:**
- Set `PIPEBOARD_API_TOKEN` environment variable for Pipeboard auth
- Or pass `meta_access_token` parameter for direct Meta API auth
## Adding New Examples
When adding new example files:
1. Include comprehensive docstrings
2. Add usage instructions in comments
3. Update this README with file descriptions
4. Follow the same authentication patterns as `http_client.py`
```
--------------------------------------------------------------------------------
/tests/README.md:
--------------------------------------------------------------------------------
```markdown
# Meta Ads MCP Tests
This directory contains integration tests for the Meta Ads MCP HTTP transport functionality.
## Test Structure
- `test_http_transport.py` - Comprehensive HTTP transport integration tests
- `conftest.py` - Pytest configuration and shared fixtures
- `__init__.py` - Python package marker
## Running Tests
### Prerequisites
1. **Start the MCP server:**
```bash
python -m meta_ads_mcp --transport streamable-http --port 8080 --host localhost
```
2. **Install test dependencies:**
```bash
pip install pytest requests
```
### Running with pytest (recommended)
```bash
# Run all tests with verbose output
python -m pytest tests/ -v
# Run specific test file
python -m pytest tests/test_http_transport.py -v
# Run with custom server URL
MCP_TEST_SERVER_URL=http://localhost:9000 python -m pytest tests/ -v
```
### Running directly
```bash
# Run the main integration test
python tests/test_http_transport.py
# Or from project root
python -m tests.test_http_transport
```
## What the Tests Validate
### ✅ HTTP Transport Layer
- Server availability and responsiveness
- JSON-RPC 2.0 protocol compliance
- Proper HTTP status codes and headers
- Request/response format validation
### ✅ MCP Protocol Compliance
- `initialize` method - Server capability exchange
- `tools/list` method - Tool discovery and enumeration
- `tools/call` method - Tool execution with parameters
- Error handling and edge cases
### ✅ Authentication Integration
- **No Authentication** - Proper rejection of unauthenticated requests
- **Pipeboard Token** - Primary authentication method (`X-PIPEBOARD-API-TOKEN`)
- **Meta App ID** - Fallback authentication method (`X-META-APP-ID`)
- **Multiple Auth Methods** - Priority handling (Pipeboard takes precedence)
### ✅ Tool Execution
- All 26 Meta Ads tools accessible via HTTP
- Authentication context properly passed to tools
- Expected behavior with test tokens (authentication required responses)
## Test Scenarios
The test suite runs multiple authentication scenarios:
1. **No Authentication**: Tests that tools properly require authentication
2. **Pipeboard Token**: Tests the primary authentication path
3. **Custom Meta App**: Tests the fallback authentication path
4. **Both Methods**: Tests authentication priority (Pipeboard preferred)
## Expected Results
With **test tokens** (used in automated tests):
- ✅ HTTP transport: All requests succeed (200 OK)
- ✅ MCP protocol: All methods work correctly
- ✅ Authentication: Headers processed and passed to tools
- ✅ Tool responses: "Authentication Required" (expected with invalid tokens)
With **real tokens** (production usage):
- ✅ All of the above PLUS actual Meta Ads data returned
## Continuous Integration
These tests are designed to be run in CI/CD pipelines:
```bash
# Start server in background
python -m meta_ads_mcp --transport streamable-http --port 8080 &
SERVER_PID=$!
# Wait for server startup
sleep 3
# Run tests
python -m pytest tests/ -v --tb=short
# Cleanup
kill $SERVER_PID
```
## Troubleshooting
**Server not running:**
```
SKIPPED [1] tests/conftest.py:25: MCP server not running at http://localhost:8080
```
→ Start the server first: `python -m meta_ads_mcp --transport streamable-http`
**Connection refused:**
```
requests.exceptions.ConnectionError: ('Connection aborted.', ...)
```
→ Check that the server is running on the expected port
**406 Not Acceptable:**
```
❌ Request failed: 406
```
→ Ensure proper Accept headers are being sent (handled automatically by test suite)
## Contributing
When adding new tests:
1. **Follow naming convention**: `test_*.py` for pytest discovery
2. **Use fixtures**: Leverage existing fixtures in `conftest.py`
3. **Test both success and failure cases**
4. **Document expected behavior** with test tokens vs real tokens
5. **Keep tests isolated**: Each test should be independent
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Meta Ads MCP
A [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) server for interacting with Meta Ads. Analyze, manage and optimize Meta advertising campaigns through an AI interface. Use an LLM to retrieve performance data, visualize ad creatives, and provide strategic insights for your ads on Facebook, Instagram, and other Meta platforms.
> **DISCLAIMER:** This is an unofficial third-party tool and is not associated with, endorsed by, or affiliated with Meta in any way. This project is maintained independently and uses Meta's public APIs according to their terms of service. Meta, Facebook, Instagram, and other Meta brand names are trademarks of their respective owners.
[](https://github.com/user-attachments/assets/3e605cee-d289-414b-814c-6299e7f3383e)
[](https://lobehub.com/mcp/nictuku-meta-ads-mcp)
mcp-name: co.pipeboard/meta-ads-mcp
## Community & Support
- [Discord](https://discord.gg/YzMwQ8zrjr). Join the community.
- [Email Support](mailto:[email protected]). Email us for support.
## Table of Contents
- [🚀 Getting started with Remote MCP (Recommended for Marketers)](#getting-started-with-remote-mcp-recommended)
- [Local Installation (Technical Users Only)](#local-installation-technical-users-only)
- [Features](#features)
- [Configuration](#configuration)
- [Available MCP Tools](#available-mcp-tools)
- [Licensing](#licensing)
- [Privacy and Security](#privacy-and-security)
- [Testing](#testing)
- [Troubleshooting](#troubleshooting)
## Getting started with Remote MCP (Recommended)
The fastest and most reliable way to get started is to **[🚀 Get started with our Meta Ads Remote MCP](https://pipeboard.co)**. Our cloud service uses streamable HTTP transport for reliable, scalable access to Meta Ads data. No technical setup required - just connect and start analyzing your ad campaigns with AI!
### For Claude Pro/Max Users
1. Go to [claude.ai/settings/integrations](https://claude.ai/settings/integrations) (requires Claude Pro or Max)
2. Click "Add Integration" and enter:
- **Name**: "Pipeboard Meta Ads" (or any name you prefer)
- **Integration URL**: `https://mcp.pipeboard.co/meta-ads-mcp`
3. Click "Connect" next to the integration and follow the prompts to:
- Login to Pipeboard
- Connect your Facebook Ads account
That's it! You can now ask Claude to analyze your Meta ad campaigns, get performance insights, and manage your advertising.
### For Cursor Users
Add the following to your `~/.cursor/mcp.json`. Once you enable the remote MCP, click on "Needs login" to finish the login process.
```json
{
"mcpServers": {
"meta-ads-remote": {
"url": "https://mcp.pipeboard.co/meta-ads-mcp"
}
}
}
```
### For Other MCP Clients
Use the Remote MCP URL: `https://mcp.pipeboard.co/meta-ads-mcp`
**[📖 Get detailed setup instructions for your AI client here](https://pipeboard.co)**
## Local Installation (Technical Users Only)
If you're a developer or need to customize the installation, you can run Meta Ads MCP locally. **Most marketers should use the Remote MCP above instead!** For complete technical setup instructions, see our **[Local Installation Guide](LOCAL_INSTALLATION.md)**.
Meta Ads MCP also supports **streamable HTTP transport**, allowing you to run it as a standalone HTTP API for web applications and custom integrations. See **[Streamable HTTP Setup Guide](STREAMABLE_HTTP_SETUP.md)** for complete instructions.
### Quick Local Setup
```bash
# Install via uvx (recommended)
uvx meta-ads-mcp
# Set your Pipeboard token
export PIPEBOARD_API_TOKEN=your_pipeboard_token
# Add to your MCP client configuration
```
For detailed step-by-step instructions, authentication setup, debugging, and troubleshooting, visit **[LOCAL_INSTALLATION.md](LOCAL_INSTALLATION.md)**.
## Features
- **AI-Powered Campaign Analysis**: Let your favorite LLM analyze your campaigns and provide actionable insights on performance
- **Strategic Recommendations**: Receive data-backed suggestions for optimizing ad spend, targeting, and creative content
- **Automated Monitoring**: Ask any MCP-compatible LLM to track performance metrics and alert you about significant changes
- **Budget Optimization**: Get recommendations for reallocating budget to better-performing ad sets
- **Creative Improvement**: Receive feedback on ad copy, imagery, and calls-to-action
- **Dynamic Creative Testing**: Easy API for both simple ads (single headline/description) and advanced A/B testing (multiple headlines/descriptions)
- **Campaign Management**: Request changes to campaigns, ad sets, and ads (all changes require explicit confirmation)
- **Cross-Platform Integration**: Works with Facebook, Instagram, and all Meta ad platforms
- **Universal LLM Support**: Compatible with any MCP client including Claude Desktop, Cursor, Cherry Studio, and more
- **Enhanced Search**: Generic search function includes page searching when queries mention "page" or "pages"
- **Simple Authentication**: Easy setup with secure OAuth authentication
- **Cross-Platform Support**: Works on Windows, macOS, and Linux
## Configuration
### Remote MCP (Recommended)
**[✨ Get started with Remote MCP here](https://pipeboard.co)** - no technical setup required! Just connect your Facebook Ads account and start asking AI to analyze your campaigns.
### Local Installation (Technical Users)
For local installation configuration, authentication options, and advanced technical setup, see our **[Local Installation Guide](LOCAL_INSTALLATION.md)**.
### Available MCP Tools
1. `mcp_meta_ads_get_ad_accounts`
- Get ad accounts accessible by a user
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `user_id`: Meta user ID or "me" for the current user
- `limit`: Maximum number of accounts to return (default: 200)
- Returns: List of accessible ad accounts with their details
2. `mcp_meta_ads_get_account_info`
- Get detailed information about a specific ad account
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- Returns: Detailed information about the specified account
3. `mcp_meta_ads_get_account_pages`
- Get pages associated with a Meta Ads account
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX) or "me" for the current user's pages
- Returns: List of pages associated with the account, useful for ad creation and management
4. `mcp_meta_ads_get_campaigns`
- Get campaigns for a Meta Ads account with optional filtering
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- `limit`: Maximum number of campaigns to return (default: 10)
- `status_filter`: Filter by status (empty for all, or 'ACTIVE', 'PAUSED', etc.)
- Returns: List of campaigns matching the criteria
5. `mcp_meta_ads_get_campaign_details`
- Get detailed information about a specific campaign
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `campaign_id`: Meta Ads campaign ID
- Returns: Detailed information about the specified campaign
6. `mcp_meta_ads_create_campaign`
- Create a new campaign in a Meta Ads account
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- `name`: Campaign name
- `objective`: Campaign objective (ODAX, outcome-based). Must be one of:
- `OUTCOME_AWARENESS`
- `OUTCOME_TRAFFIC`
- `OUTCOME_ENGAGEMENT`
- `OUTCOME_LEADS`
- `OUTCOME_SALES`
- `OUTCOME_APP_PROMOTION`
Note: Legacy objectives such as `BRAND_AWARENESS`, `LINK_CLICKS`, `CONVERSIONS`, `APP_INSTALLS`, etc. are no longer valid for new campaigns and will cause a 400 error. Use the outcome-based values above. Common mappings:
- `BRAND_AWARENESS` → `OUTCOME_AWARENESS`
- `REACH` → `OUTCOME_AWARENESS`
- `LINK_CLICKS`, `TRAFFIC` → `OUTCOME_TRAFFIC`
- `POST_ENGAGEMENT`, `PAGE_LIKES`, `EVENT_RESPONSES`, `VIDEO_VIEWS` → `OUTCOME_ENGAGEMENT`
- `LEAD_GENERATION` → `OUTCOME_LEADS`
- `CONVERSIONS`, `CATALOG_SALES`, `MESSAGES` (sales-focused flows) → `OUTCOME_SALES`
- `APP_INSTALLS` → `OUTCOME_APP_PROMOTION`
- `status`: Initial campaign status (default: PAUSED)
- `special_ad_categories`: List of special ad categories if applicable
- `daily_budget`: Daily budget in account currency (in cents)
- `lifetime_budget`: Lifetime budget in account currency (in cents)
- `bid_strategy`: Bid strategy. Must be one of: `LOWEST_COST_WITHOUT_CAP`, `LOWEST_COST_WITH_BID_CAP`, `COST_CAP`, `LOWEST_COST_WITH_MIN_ROAS`.
- Returns: Confirmation with new campaign details
- Example:
```json
{
"name": "2025 - Bedroom Furniture - Awareness",
"account_id": "act_123456789012345",
"objective": "OUTCOME_AWARENESS",
"special_ad_categories": [],
"status": "PAUSED",
"buying_type": "AUCTION",
"bid_strategy": "LOWEST_COST_WITHOUT_CAP",
"daily_budget": 10000
}
```
7. `mcp_meta_ads_get_adsets`
- Get ad sets for a Meta Ads account with optional filtering by campaign
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- `limit`: Maximum number of ad sets to return (default: 10)
- `campaign_id`: Optional campaign ID to filter by
- Returns: List of ad sets matching the criteria
8. `mcp_meta_ads_get_adset_details`
- Get detailed information about a specific ad set
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `adset_id`: Meta Ads ad set ID
- Returns: Detailed information about the specified ad set
9. `mcp_meta_ads_create_adset`
- Create a new ad set in a Meta Ads account
- Inputs:
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- `campaign_id`: Meta Ads campaign ID this ad set belongs to
- `name`: Ad set name
- `status`: Initial ad set status (default: PAUSED)
- `daily_budget`: Daily budget in account currency (in cents) as a string
- `lifetime_budget`: Lifetime budget in account currency (in cents) as a string
- `targeting`: Targeting specifications (e.g., age, location, interests)
- `optimization_goal`: Conversion optimization goal (e.g., 'LINK_CLICKS')
- `billing_event`: How you're charged (e.g., 'IMPRESSIONS')
- `bid_amount`: Bid amount in account currency (in cents)
- `bid_strategy`: Bid strategy (e.g., 'LOWEST_COST')
- `start_time`, `end_time`: Optional start/end times (ISO 8601)
- `access_token` (optional): Meta API access token
- Returns: Confirmation with new ad set details
10. `mcp_meta_ads_get_ads`
- Get ads for a Meta Ads account with optional filtering
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- `limit`: Maximum number of ads to return (default: 10)
- `campaign_id`: Optional campaign ID to filter by
- `adset_id`: Optional ad set ID to filter by
- Returns: List of ads matching the criteria
11. `mcp_meta_ads_create_ad`
- Create a new ad with an existing creative
- Inputs:
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- `name`: Ad name
- `adset_id`: Ad set ID where this ad will be placed
- `creative_id`: ID of an existing creative to use
- `status`: Initial ad status (default: PAUSED)
- `bid_amount`: Optional bid amount (in cents)
- `tracking_specs`: Optional tracking specifications
- `access_token` (optional): Meta API access token
- Returns: Confirmation with new ad details
12. `mcp_meta_ads_get_ad_details`
- Get detailed information about a specific ad
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `ad_id`: Meta Ads ad ID
- Returns: Detailed information about the specified ad
13. `mcp_meta_ads_get_ad_creatives`
- Get creative details for a specific ad
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `ad_id`: Meta Ads ad ID
- Returns: Creative details including text, images, and URLs
14. `mcp_meta_ads_create_ad_creative`
- Create a new ad creative using an uploaded image hash
- Inputs:
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- `name`: Creative name
- `image_hash`: Hash of the uploaded image
- `page_id`: Facebook Page ID for the ad
- `link_url`: Destination URL
- `message`: Ad copy/text
- `headline`: Single headline for simple ads (cannot be used with headlines)
- `headlines`: List of headlines for dynamic creative testing (cannot be used with headline)
- `description`: Single description for simple ads (cannot be used with descriptions)
- `descriptions`: List of descriptions for dynamic creative testing (cannot be used with description)
- `dynamic_creative_spec`: Dynamic creative optimization settings
- `call_to_action_type`: CTA button type (e.g., 'LEARN_MORE')
- `instagram_actor_id`: Optional Instagram account ID
- `access_token` (optional): Meta API access token
- Returns: Confirmation with new creative details
15. `mcp_meta_ads_update_ad_creative`
- Update an existing ad creative with new content or settings
- Inputs:
- `creative_id`: Meta Ads creative ID to update
- `name`: New creative name
- `message`: New ad copy/text
- `headline`: Single headline for simple ads (cannot be used with headlines)
- `headlines`: New list of headlines for dynamic creative testing (cannot be used with headline)
- `description`: Single description for simple ads (cannot be used with descriptions)
- `descriptions`: New list of descriptions for dynamic creative testing (cannot be used with description)
- `dynamic_creative_spec`: New dynamic creative optimization settings
- `call_to_action_type`: New call to action button type
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- Returns: Confirmation with updated creative details
16. `mcp_meta_ads_upload_ad_image`
- Upload an image to use in Meta Ads creatives
- Inputs:
- `account_id`: Meta Ads account ID (format: act_XXXXXXXXX)
- `image_path`: Path to the image file to upload
- `name`: Optional name for the image
- `access_token` (optional): Meta API access token
- Returns: JSON response with image details including hash
17. `mcp_meta_ads_get_ad_image`
- Get, download, and visualize a Meta ad image in one step
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `ad_id`: Meta Ads ad ID
- Returns: The ad image ready for direct visual analysis
18. `mcp_meta_ads_update_ad`
- Update an ad with new settings
- Inputs:
- `ad_id`: Meta Ads ad ID
- `status`: Update ad status (ACTIVE, PAUSED, etc.)
- `bid_amount`: Bid amount in account currency (in cents for USD)
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- Returns: Confirmation with updated ad details and a confirmation link
19. `mcp_meta_ads_update_adset`
- Update an ad set with new settings including frequency caps
- Inputs:
- `adset_id`: Meta Ads ad set ID
- `frequency_control_specs`: List of frequency control specifications
- `bid_strategy`: Bid strategy (e.g., 'LOWEST_COST_WITH_BID_CAP')
- `bid_amount`: Bid amount in account currency (in cents for USD)
- `status`: Update ad set status (ACTIVE, PAUSED, etc.)
- `targeting`: Targeting specifications including targeting_automation
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- Returns: Confirmation with updated ad set details and a confirmation link
20. `mcp_meta_ads_get_insights`
- Get performance insights for a campaign, ad set, ad or account
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `object_id`: ID of the campaign, ad set, ad or account
- `time_range`: Time range for insights (default: maximum)
- `breakdown`: Optional breakdown dimension (e.g., age, gender, country)
- `level`: Level of aggregation (ad, adset, campaign, account)
- Returns: Performance metrics for the specified object
21. `mcp_meta_ads_get_login_link`
- Get a clickable login link for Meta Ads authentication
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- Returns: A clickable resource link for Meta authentication
22. `mcp_meta_ads_create_budget_schedule`
- Create a budget schedule for a Meta Ads campaign
- Inputs:
- `campaign_id`: Meta Ads campaign ID
- `budget_value`: Amount of budget increase
- `budget_value_type`: Type of budget value ("ABSOLUTE" or "MULTIPLIER")
- `time_start`: Unix timestamp for when the high demand period should start
- `time_end`: Unix timestamp for when the high demand period should end
- `access_token` (optional): Meta API access token
- Returns: JSON string with the ID of the created budget schedule or an error message
23. `mcp_meta_ads_search_interests`
- Search for interest targeting options by keyword
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `query`: Search term for interests (e.g., "baseball", "cooking", "travel")
- `limit`: Maximum number of results to return (default: 25)
- Returns: Interest data with id, name, audience_size, and path fields
24. `mcp_meta_ads_get_interest_suggestions`
- Get interest suggestions based on existing interests
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `interest_list`: List of interest names to get suggestions for (e.g., ["Basketball", "Soccer"])
- `limit`: Maximum number of suggestions to return (default: 25)
- Returns: Suggested interests with id, name, audience_size, and description fields
25. `mcp_meta_ads_validate_interests`
- Validate interest names or IDs for targeting
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `interest_list`: List of interest names to validate (e.g., ["Japan", "Basketball"])
- `interest_fbid_list`: List of interest IDs to validate (e.g., ["6003700426513"])
- Returns: Validation results showing valid status and audience_size for each interest
26. `mcp_meta_ads_search_behaviors`
- Get all available behavior targeting options
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `limit`: Maximum number of results to return (default: 50)
- Returns: Behavior targeting options with id, name, audience_size bounds, path, and description
27. `mcp_meta_ads_search_demographics`
- Get demographic targeting options
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `demographic_class`: Type of demographics ('demographics', 'life_events', 'industries', 'income', 'family_statuses', 'user_device', 'user_os')
- `limit`: Maximum number of results to return (default: 50)
- Returns: Demographic targeting options with id, name, audience_size bounds, path, and description
28. `mcp_meta_ads_search_geo_locations`
- Search for geographic targeting locations
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `query`: Search term for locations (e.g., "New York", "California", "Japan")
- `location_types`: Types of locations to search (['country', 'region', 'city', 'zip', 'geo_market', 'electoral_district'])
- `limit`: Maximum number of results to return (default: 25)
- Returns: Location data with key, name, type, and geographic hierarchy information
29. `mcp_meta_ads_search` (Enhanced)
- Generic search across accounts, campaigns, ads, and pages
- Automatically includes page searching when query mentions "page" or "pages"
- Inputs:
- `access_token` (optional): Meta API access token (will use cached token if not provided)
- `query`: Search query string (e.g., "Injury Payouts pages", "active campaigns")
- Returns: List of matching record IDs in ChatGPT-compatible format
## Licensing
Meta Ads MCP is licensed under the [Business Source License 1.1](LICENSE), which means:
- ✅ **Free to use** for individual and business purposes
- ✅ **Modify and customize** as needed
- ✅ **Redistribute** to others
- ✅ **Becomes fully open source** (Apache 2.0) on January 1, 2029
The only restriction is that you cannot offer this as a competing hosted service. For questions about commercial licensing, please contact us.
## Privacy and Security
Meta Ads MCP follows security best practices with secure token management and automatic authentication handling.
- **Remote MCP**: All authentication is handled securely in the cloud - no local token storage required
- **Local Installation**: Tokens are cached securely on your local machine - see [Local Installation Guide](LOCAL_INSTALLATION.md) for details
## Testing
### Basic Testing
Test your Meta Ads MCP connection with any MCP client:
1. **Verify Account Access**: Ask your LLM to use `mcp_meta_ads_get_ad_accounts`
2. **Check Account Details**: Use `mcp_meta_ads_get_account_info` with your account ID
3. **List Campaigns**: Try `mcp_meta_ads_get_campaigns` to see your ad campaigns
For detailed local installation testing, see [Local Installation Guide](LOCAL_INSTALLATION.md).
## Troubleshooting
### 💡 Quick Fix: Skip the Technical Setup!
The easiest way to avoid any setup issues is to **[🎯 use our Remote MCP instead](https://pipeboard.co)**. No downloads, no configuration - just connect your ads account and start getting AI insights on your campaigns immediately!
### Local Installation Issues
For comprehensive troubleshooting, debugging, and local installation issues, see our **[Local Installation Guide](LOCAL_INSTALLATION.md)** which includes:
- Authentication troubleshooting
- Installation issues and solutions
- API error resolution
- Debug logs and diagnostic commands
- Performance optimization tips
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
# Tests package for Meta Ads MCP
```
--------------------------------------------------------------------------------
/setup.py:
--------------------------------------------------------------------------------
```python
"""Setup script for meta-ads-mcp package."""
from setuptools import setup
if __name__ == "__main__":
setup()
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
httpx>=0.26.0
mcp[cli]==1.12.2
python-dotenv>=1.1.0
requests>=2.32.3
Pillow>=10.0.0
pathlib>=1.0.1
python-dateutil>=2.8.2
```
--------------------------------------------------------------------------------
/meta_ads_mcp/__main__.py:
--------------------------------------------------------------------------------
```python
"""
Meta Ads MCP - Main Entry Point
This module allows the package to be executed directly via `python -m meta_ads_mcp`
"""
from meta_ads_mcp.core.server import main
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/future_improvements.md:
--------------------------------------------------------------------------------
```markdown
# Future Improvements for Meta Ads MCP
## Note about Meta Ads development work
If you update the MCP server code, please note that *I* have to restart the MCP server. After the server code is changed, ask me to restart it and then proceed with your testing after I confirm it's restarted.
## Access token should remain internal only
Don't share it ever with the LLM, only update the auth cache.
Future improvements can be added to this file as needed.
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
FROM python:3.11-slim
# Install system dependencies
RUN apt-get update && \
apt-get install -y --no-install-recommends gcc && \
rm -rf /var/lib/apt/lists/*
# Set working directory
WORKDIR /app
# Install uv
RUN pip install --upgrade pip && \
pip install uv
# Copy requirements file
COPY requirements.txt .
# Install dependencies using uv with --system flag
RUN uv pip install --system -r requirements.txt
# Copy the rest of the application
COPY . .
# Command to run the Meta Ads MCP server
CMD ["python", "-m", "meta_ads_mcp"]
```
--------------------------------------------------------------------------------
/server.json:
--------------------------------------------------------------------------------
```json
{
"$schema": "https://static.modelcontextprotocol.io/schemas/2025-09-16/server.schema.json",
"name": "co.pipeboard/meta-ads-mcp",
"description": "Facebook / Meta Ads automation with AI: analyze performance, test creatives, optimize spend.",
"version": "1.0.15",
"remotes": [
{
"type": "streamable-http",
"url": "https://mcp.pipeboard.co/meta-ads-mcp"
}
],
"packages": [
{
"registryType": "pypi",
"identifier": "meta-ads-mcp",
"version": "1.0.15",
"transport": {
"type": "stdio"
}
}
]
}
```
--------------------------------------------------------------------------------
/tests/test_openai.py:
--------------------------------------------------------------------------------
```python
import os
import pytest
# Skip this test entirely if the optional 'openai' dependency is not installed
openai = pytest.importorskip("openai", reason="openai package not installed")
@pytest.mark.skipif(
not os.getenv("PIPEBOARD_API_TOKEN"),
reason="PIPEBOARD_API_TOKEN not set - skipping OpenAI integration test"
)
def test_openai_mcp_integration():
"""Test OpenAI integration with Meta Ads MCP via Pipeboard."""
client = openai.OpenAI()
resp = client.responses.create(
model="gpt-4.1",
tools=[{
"type": "mcp",
"server_label": "meta-ads",
"server_url": "https://mcp.pipeboard.co/meta-ads-mcp",
"headers": {
"Authorization": f"Bearer {os.getenv('PIPEBOARD_API_TOKEN')}"
},
"require_approval": "never",
}],
input="What are my meta ad accounts? Do not pass access_token since auth is already done.",
)
assert resp.output_text is not None
print(resp.output_text)
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
startCommand:
type: stdio
configSchema:
type: object
properties:
pipeboardApiToken:
type: string
description: "Pipeboard API token for Meta authentication (recommended). Get your free token at https://pipeboard.co"
metaAppId:
type: string
description: "Meta App ID (Client ID) for direct OAuth method (only needed if not using Pipeboard authentication)"
required: []
commandFunction: |
(config) => {
const env = {};
const args = ["-m", "meta_ads_mcp"];
// Add Pipeboard API token to environment if provided (recommended auth method)
if (config.pipeboardApiToken) {
env.PIPEBOARD_API_TOKEN = config.pipeboardApiToken;
}
// Add Meta App ID as command-line argument if provided (alternative auth method)
if (config.metaAppId) {
args.push("--app-id", config.metaAppId);
}
return {
command: 'python',
args: args,
env: env
};
}
remotes:
- type: streamable-http
url: "https://mcp.pipeboard.co/meta-ads-mcp"
```
--------------------------------------------------------------------------------
/.github/workflows/publish-mcp.yml:
--------------------------------------------------------------------------------
```yaml
name: Publish to MCP Registry (manual)
on:
# This workflow is kept for manual runs only. The normal release flow
# is handled by the consolidated release workflow.
workflow_dispatch:
jobs:
publish:
runs-on: ubuntu-latest
permissions:
id-token: write
contents: read
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Install MCP Publisher
run: |
curl -L "https://github.com/modelcontextprotocol/registry/releases/download/v1.1.0/mcp-publisher_1.1.0_$(uname -s | tr '[:upper:]' '[:lower:]')_$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/').tar.gz" | tar xz mcp-publisher
- name: Login to MCP Registry (DNS auth)
run: |
echo "${{ secrets.MCP_PRIVATE_KEY }}" > temp_key.pem
PRIVATE_KEY_HEX=$(openssl pkey -in temp_key.pem -noout -text | grep -A3 "priv:" | tail -n +2 | tr -d ' :\n')
./mcp-publisher login dns --domain pipeboard.co --private-key "$PRIVATE_KEY_HEX"
rm -f temp_key.pem
- name: Publish to MCP Registry
run: ./mcp-publisher publish
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/resources.py:
--------------------------------------------------------------------------------
```python
"""Resource handling for Meta Ads API."""
from typing import Dict, Any
import base64
from .utils import ad_creative_images
async def list_resources() -> Dict[str, Any]:
"""
List all available resources (like ad creative images)
Returns:
Dictionary with resources list
"""
resources = []
# Add all ad creative images as resources
for resource_id, image_info in ad_creative_images.items():
resources.append({
"uri": f"meta-ads://images/{resource_id}",
"mimeType": image_info["mime_type"],
"name": image_info["name"]
})
return {"resources": resources}
async def get_resource(resource_id: str) -> Dict[str, Any]:
"""
Get a specific resource by URI
Args:
resource_id: Unique identifier for the resource
Returns:
Dictionary with resource data
"""
if resource_id in ad_creative_images:
image_info = ad_creative_images[resource_id]
return {
"data": base64.b64encode(image_info["data"]).decode("utf-8"),
"mimeType": image_info["mime_type"]
}
# Resource not found
return {"error": f"Resource not found: {resource_id}"}
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "meta-ads-mcp"
version = "1.0.15"
description = "Model Context Protocol (MCP) server for interacting with Meta Ads API"
readme = "README.md"
requires-python = ">=3.10"
authors = [
{name = "Yves Junqueira", email = "[email protected]"},
]
keywords = ["meta", "facebook", "ads", "api", "mcp", "claude"]
license = {text = "BUSL-1.1"}
classifiers = [
"Programming Language :: Python :: 3",
"License :: Other/Proprietary License",
"Operating System :: OS Independent",
]
dependencies = [
"httpx>=0.26.0",
"mcp[cli]==1.12.2",
"python-dotenv>=1.1.0",
"requests>=2.32.3",
"Pillow>=10.0.0",
"pathlib>=1.0.1",
"python-dateutil>=2.8.2",
"pytest>=8.4.1",
"pytest-asyncio>=1.0.0",
]
[project.urls]
"Homepage" = "https://github.com/pipeboard-co/meta-ads-mcp"
"Bug Tracker" = "https://github.com/pipeboard-co/meta-ads-mcp/issues"
[project.scripts]
meta-ads-mcp = "meta_ads_mcp:entrypoint"
[tool.hatch.build.targets.wheel]
packages = ["meta_ads_mcp"]
[tool.pytest.ini_options]
markers = [
"e2e: marks tests as end-to-end (requires running MCP server) - excluded from default runs",
]
addopts = "-v --strict-markers -m 'not e2e'"
testpaths = ["tests"]
asyncio_mode = "auto"
```
--------------------------------------------------------------------------------
/.github/workflows/test.yml:
--------------------------------------------------------------------------------
```yaml
name: Test and Build
on:
push:
branches: [ main, master ]
pull_request:
branches: [ main, master ]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.10", "3.11", "3.12"]
steps:
- name: Check out code
uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install build pytest
pip install -e .
- name: Test package build
run: python -m build
- name: Test package installation
run: |
pip install dist/*.whl
python -c "import meta_ads_mcp; print('Package imported successfully')"
validate-version:
runs-on: ubuntu-latest
if: github.event_name == 'pull_request'
steps:
- name: Check out code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.10"
- name: Check version bump
run: |
# This is a simple check - you might want to make it more sophisticated
echo "Current version in pyproject.toml:"
grep "version = " pyproject.toml
```
--------------------------------------------------------------------------------
/meta_ads_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
"""
Meta Ads MCP - Python Package
This package provides a Meta Ads MCP integration
"""
from meta_ads_mcp.core.server import main
__version__ = "1.0.15"
__all__ = [
'get_ad_accounts',
'get_account_info',
'get_campaigns',
'get_campaign_details',
'create_campaign',
'get_adsets',
'get_adset_details',
'update_adset',
'get_ads',
'get_ad_details',
'get_ad_creatives',
'get_ad_image',
'update_ad',
'get_insights',
# 'get_login_link' is conditionally exported via core.__all__
'login_cli',
'main',
'search_interests',
'get_interest_suggestions',
'estimate_audience_size',
'search_behaviors',
'search_demographics',
'search_geo_locations'
]
# Import key functions to make them available at package level
from .core import (
get_ad_accounts,
get_account_info,
get_campaigns,
get_campaign_details,
create_campaign,
get_adsets,
get_adset_details,
update_adset,
get_ads,
get_ad_details,
get_ad_creatives,
get_ad_image,
update_ad,
get_insights,
login_cli,
main,
search_interests,
get_interest_suggestions,
estimate_audience_size,
search_behaviors,
search_demographics,
search_geo_locations
)
# Define a main function to be used as a package entry point
def entrypoint():
"""Main entry point for the package when invoked with uvx."""
return main()
# Re-export main for direct access
main = main
```
--------------------------------------------------------------------------------
/META_API_NOTES.md:
--------------------------------------------------------------------------------
```markdown
# Meta Ads API Notes and Limitations
## Frequency Cap Visibility
The Meta Marketing API has some limitations regarding frequency cap visibility:
1. **Optimization Goal Dependency**: Frequency cap settings (`frequency_control_specs`) are only visible in API responses for ad sets where the optimization goal is set to REACH. For other optimization goals (like LINK_CLICKS, CONVERSIONS, etc.), the frequency caps will still work but won't be visible through the API.
2. **Verifying Frequency Caps**: Since frequency caps may not be directly visible, you can verify they're working by monitoring:
- The frequency metric in ad insights
- The ratio between reach and impressions over time
- The actual frequency cap behavior in the Meta Ads Manager UI
## Other API Behaviors to Note
1. **Field Visibility**: Some fields may not appear in API responses even when explicitly requested. This doesn't necessarily mean the field isn't set - it may just not be visible through the API.
2. **Response Filtering**: The API may filter out empty or default values from responses to reduce payload size. If a field is missing from a response, it might mean:
- The field is not set
- The field has a default value
- The field is not applicable for the current configuration
3. **Best Practices**:
- Always verify important changes through both the API and Meta Ads Manager UI
- Use insights and metrics to confirm behavioral changes when direct field access is limited
- Consider the optimization goal when setting up features like frequency caps
## Updates and Changes
Meta frequently updates their API behavior. These notes will be updated as we discover new limitations or changes in the API's behavior.
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
"""
Pytest configuration for Meta Ads MCP tests
This file provides common fixtures and configuration for all tests.
"""
import pytest
import requests
import time
import os
@pytest.fixture(scope="session")
def server_url():
"""Default server URL for tests"""
return os.environ.get("MCP_TEST_SERVER_URL", "http://localhost:8080")
@pytest.fixture(scope="session")
def check_server_running(server_url):
"""
Check if the MCP server is running before running tests.
This fixture will skip tests if the server is not available.
"""
try:
response = requests.get(f"{server_url}/", timeout=5)
# We expect 404 for root path, but it means server is running
if response.status_code not in [200, 404]:
pytest.skip(f"MCP server not responding correctly at {server_url}")
return True
except requests.exceptions.RequestException:
pytest.skip(
f"MCP server not running at {server_url}. "
f"Start with: python -m meta_ads_mcp --transport streamable-http"
)
@pytest.fixture
def test_headers():
"""Common test headers for HTTP requests"""
return {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
"User-Agent": "MCP-Test-Client/1.0"
}
@pytest.fixture
def pipeboard_auth_headers(test_headers):
"""Headers with Pipeboard authentication token"""
headers = test_headers.copy()
headers["Authorization"] = "Bearer test_pipeboard_token_12345"
return headers
@pytest.fixture
def meta_app_auth_headers(test_headers):
"""Headers with Meta app ID authentication"""
headers = test_headers.copy()
headers["X-META-APP-ID"] = "123456789012345"
return headers
```
--------------------------------------------------------------------------------
/tests/test_create_simple_creative_e2e.py:
--------------------------------------------------------------------------------
```python
"""End-to-end test for creating simple creatives with singular headline/description."""
import pytest
import json
import os
from meta_ads_mcp.core.ads import create_ad_creative
@pytest.mark.skip(reason="Requires authentication - run manually with: pytest tests/test_create_simple_creative_e2e.py -v")
@pytest.mark.asyncio
async def test_create_simple_creative_with_real_api():
"""Test creating a simple creative with singular headline/description using real Meta API."""
# Account and image details from user
account_id = "act_3182643988557192"
image_hash = "ca228ac8ff3a66dca9435c90dd6953d6"
# Create a simple creative with singular headline and description
result = await create_ad_creative(
account_id=account_id,
image_hash=image_hash,
name="E2E Test - Simple Creative",
link_url="https://example.com/",
message="This is a test message for the ad.",
headline="Test Headline",
description="Test description for ad.",
call_to_action_type="LEARN_MORE"
)
print("\n=== API Response ===")
print(result)
result_data = json.loads(result)
# Check if there's an error
if "error" in result_data:
pytest.fail(f"Creative creation failed: {result_data['error']}")
# Verify success
assert "success" in result_data or "creative_id" in result_data or "id" in result_data, \
f"Expected success response, got: {result_data}"
print("\n✅ Simple creative created successfully!")
if "creative_id" in result_data:
print(f"Creative ID: {result_data['creative_id']}")
elif "details" in result_data and "id" in result_data["details"]:
print(f"Creative ID: {result_data['details']['id']}")
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/__init__.py:
--------------------------------------------------------------------------------
```python
"""Core functionality for Meta Ads API MCP package."""
from .server import mcp_server
from .accounts import get_ad_accounts, get_account_info
from .campaigns import get_campaigns, get_campaign_details, create_campaign
from .adsets import get_adsets, get_adset_details, update_adset
from .ads import get_ads, get_ad_details, get_ad_creatives, get_ad_image, update_ad
from .insights import get_insights
from . import authentication # Import module to register conditional auth tools
from .server import login_cli, main
from .auth import login
from . import ads_library # Import module to register conditional tools
from .budget_schedules import create_budget_schedule
from .targeting import search_interests, get_interest_suggestions, estimate_audience_size, search_behaviors, search_demographics, search_geo_locations
from . import reports # Import module to register conditional tools
from . import duplication # Import module to register conditional duplication tools
from .openai_deep_research import search, fetch # OpenAI MCP Deep Research tools
__all__ = [
'mcp_server',
'get_ad_accounts',
'get_account_info',
'get_campaigns',
'get_campaign_details',
'create_campaign',
'get_adsets',
'get_adset_details',
'update_adset',
'get_ads',
'get_ad_details',
'get_ad_creatives',
'get_ad_image',
'update_ad',
'get_insights',
# Note: 'get_login_link' is registered conditionally by the authentication module
'login_cli',
'login',
'main',
'create_budget_schedule',
'search_interests',
'get_interest_suggestions',
'estimate_audience_size',
'search_behaviors',
'search_demographics',
'search_geo_locations',
'search', # OpenAI MCP Deep Research search tool
'fetch', # OpenAI MCP Deep Research fetch tool
]
```
--------------------------------------------------------------------------------
/meta_ads_auth.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
echo "Starting Meta Ads MCP with Pipeboard authentication..."
# Set the Pipeboard API token as an environment variable
export PIPEBOARD_API_TOKEN="pk_8ee8c727644d4d32b646ddcf16f2385e"
# Check if the Pipeboard server is running and can handle meta auth
echo "Checking if Pipeboard meta auth endpoint is available..."
curl -s -o /dev/null -w "%{http_code}" "http://localhost:3000/api/meta" > /dev/null
if [ $? -ne 0 ]; then
echo "Warning: Could not connect to Pipeboard meta endpoint"
echo "Please make sure the Pipeboard server is running before proceeding."
echo "You can start it with: cd /path/to/pipeboard && npm run dev"
echo ""
echo "As a test, you can try running this command separately:"
echo "curl -X POST \"http://localhost:3000/api/meta/auth?api_token=pk_8ee8c727644d4d32b646ddcf16f2385e\" -H \"Content-Type: application/json\""
echo ""
read -p "Press Enter to continue anyway or Ctrl+C to abort..."
fi
# Try direct auth to test the endpoint
echo "Testing direct authentication with Pipeboard..."
AUTH_RESPONSE=$(curl -s -X POST "http://localhost:3000/api/meta/auth?api_token=pk_8ee8c727644d4d32b646ddcf16f2385e" -H "Content-Type: application/json")
if [[ $AUTH_RESPONSE == *"loginUrl"* ]]; then
echo "Authentication endpoint working correctly!"
LOGIN_URL=$(echo $AUTH_RESPONSE | grep -o 'https://[^"]*')
echo "Login URL: $LOGIN_URL"
# Open the browser directly as a fallback
echo "Opening browser to login URL (as a fallback)..."
if [ "$(uname)" == "Darwin" ]; then
# macOS
open "$LOGIN_URL"
elif [ "$(expr substr $(uname -s) 1 5)" == "Linux" ]; then
# Linux
xdg-open "$LOGIN_URL" || firefox "$LOGIN_URL" || google-chrome "$LOGIN_URL" || echo "Could not open browser automatically"
else
# Windows or others
echo "Please open this URL in your browser manually: $LOGIN_URL"
fi
echo "After authorizing in your browser, the MCP script will retrieve the token."
else
echo "Warning: Authentication endpoint test failed!"
echo "Response: $AUTH_RESPONSE"
read -p "Press Enter to continue anyway or Ctrl+C to abort..."
fi
# Run the meta-ads-mcp package
echo "Running Meta Ads MCP..."
python -m meta_ads_mcp
echo "Meta Ads MCP server is now running."
echo "If you had to manually authenticate, the token should be cached now."
```
--------------------------------------------------------------------------------
/tests/test_is_dynamic_creative_adset.py:
--------------------------------------------------------------------------------
```python
import json
import pytest
from unittest.mock import AsyncMock, patch
from meta_ads_mcp.core.adsets import create_adset, update_adset, get_adsets, get_adset_details
@pytest.mark.asyncio
async def test_create_adset_includes_is_dynamic_creative_true():
sample_response = {"id": "adset_1", "name": "DC Adset"}
with patch('meta_ads_mcp.core.adsets.make_api_request', new_callable=AsyncMock) as mock_api:
mock_api.return_value = sample_response
result = await create_adset(
account_id="act_123",
campaign_id="cmp_1",
name="DC Adset",
optimization_goal="LINK_CLICKS",
billing_event="IMPRESSIONS",
targeting={"geo_locations": {"countries": ["US"]}},
is_dynamic_creative=True,
access_token="test_token",
)
assert json.loads(result)["id"] == "adset_1"
# Verify param was sent as string boolean
call_args = mock_api.call_args
params = call_args[0][2]
assert params["is_dynamic_creative"] == "true"
@pytest.mark.asyncio
async def test_update_adset_includes_is_dynamic_creative_false():
sample_response = {"success": True}
with patch('meta_ads_mcp.core.adsets.make_api_request', new_callable=AsyncMock) as mock_api:
mock_api.return_value = sample_response
result = await update_adset(
adset_id="120",
is_dynamic_creative=False,
access_token="test_token",
)
assert json.loads(result)["success"] is True
call_args = mock_api.call_args
params = call_args[0][2]
assert params["is_dynamic_creative"] == "false"
@pytest.mark.asyncio
async def test_get_adsets_fields_include_is_dynamic_creative():
sample_response = {"data": []}
with patch('meta_ads_mcp.core.adsets.make_api_request', new_callable=AsyncMock) as mock_api:
mock_api.return_value = sample_response
result = await get_adsets(account_id="act_123", access_token="test_token", limit=1)
assert json.loads(result)["data"] == []
call_args = mock_api.call_args
params = call_args[0][2]
assert "is_dynamic_creative" in params.get("fields", "")
@pytest.mark.asyncio
async def test_get_adset_details_fields_include_is_dynamic_creative():
sample_response = {"id": "120", "name": "Test", "is_dynamic_creative": True}
with patch('meta_ads_mcp.core.adsets.make_api_request', new_callable=AsyncMock) as mock_api:
mock_api.return_value = sample_response
result = await get_adset_details(adset_id="120", access_token="test_token")
assert json.loads(result)["id"] == "120"
call_args = mock_api.call_args
params = call_args[0][2]
assert "is_dynamic_creative" in params.get("fields", "")
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/budget_schedules.py:
--------------------------------------------------------------------------------
```python
"""Budget Schedule-related functionality for Meta Ads API."""
import json
from typing import Optional, Dict, Any
from .api import meta_api_tool, make_api_request
from .server import mcp_server
# Assuming no other specific dependencies from adsets.py are needed for this single function.
# If other utilities from adsets.py (like get_ad_accounts) were needed, they'd be imported here.
@mcp_server.tool()
@meta_api_tool
async def create_budget_schedule(
campaign_id: str,
budget_value: int,
budget_value_type: str,
time_start: int,
time_end: int,
access_token: Optional[str] = None
) -> str:
"""
Create a budget schedule for a Meta Ads campaign.
Allows scheduling budget increases based on anticipated high-demand periods.
The times should be provided as Unix timestamps.
Args:
campaign_id: Meta Ads campaign ID.
budget_value: Amount of budget increase. Interpreted based on budget_value_type.
budget_value_type: Type of budget value - "ABSOLUTE" or "MULTIPLIER".
time_start: Unix timestamp for when the high demand period should start.
time_end: Unix timestamp for when the high demand period should end.
access_token: Meta API access token (optional - will use cached token if not provided).
Returns:
A JSON string containing the ID of the created budget schedule or an error message.
"""
if not campaign_id:
return json.dumps({"error": "Campaign ID is required"}, indent=2)
if budget_value is None: # Check for None explicitly
return json.dumps({"error": "Budget value is required"}, indent=2)
if not budget_value_type:
return json.dumps({"error": "Budget value type is required"}, indent=2)
if budget_value_type not in ["ABSOLUTE", "MULTIPLIER"]:
return json.dumps({"error": "Invalid budget_value_type. Must be ABSOLUTE or MULTIPLIER"}, indent=2)
if time_start is None: # Check for None explicitly to allow 0
return json.dumps({"error": "Time start is required"}, indent=2)
if time_end is None: # Check for None explicitly to allow 0
return json.dumps({"error": "Time end is required"}, indent=2)
endpoint = f"{campaign_id}/budget_schedules"
params = {
"budget_value": budget_value,
"budget_value_type": budget_value_type,
"time_start": time_start,
"time_end": time_end,
}
try:
data = await make_api_request(endpoint, access_token, params, method="POST")
return json.dumps(data, indent=2)
except Exception as e:
error_msg = str(e)
# Include details about the error and the parameters sent for easier debugging
return json.dumps({
"error": "Failed to create budget schedule",
"details": error_msg,
"campaign_id": campaign_id,
"params_sent": params
}, indent=2)
```
--------------------------------------------------------------------------------
/.github/workflows/publish.yml:
--------------------------------------------------------------------------------
```yaml
name: "Release: Test, PyPI, MCP"
on:
release:
types: [published]
# Allow manual triggering for testing
workflow_dispatch:
jobs:
test_and_build:
name: Test and Build (pre-release gate)
runs-on: ubuntu-latest
permissions:
contents: read
steps:
- name: Check out code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install uv
uses: astral-sh/setup-uv@v5
- name: Install dependencies
run: |
uv sync --all-extras --dev
- name: Run tests
run: |
uv run pytest -q
- name: Build wheel and sdist
run: |
uv build
- name: Validate server.json against schema
run: |
uv run python - <<'PY'
import json, sys, urllib.request
from jsonschema import validate
from jsonschema.exceptions import ValidationError
server = json.load(open('server.json'))
schema_url = server.get('$schema')
with urllib.request.urlopen(schema_url) as r:
schema = json.load(r)
try:
validate(instance=server, schema=schema)
except ValidationError as e:
print('Schema validation failed:', e, file=sys.stderr)
sys.exit(1)
print('server.json is valid')
PY
publish_pypi:
name: Publish to PyPI
needs: test_and_build
runs-on: ubuntu-latest
environment: release
permissions:
id-token: write # IMPORTANT: this permission is mandatory for trusted publishing
contents: read
steps:
- name: Check out code
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.10"
- name: Install build dependencies
run: |
python -m pip install --upgrade pip
pip install build
- name: Build package
run: python -m build
- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
verbose: true
publish_mcp:
name: Publish to MCP Registry
needs: publish_pypi
runs-on: ubuntu-latest
permissions:
id-token: write
contents: read
steps:
- name: Check out code
uses: actions/checkout@v4
- name: Install MCP Publisher
run: |
curl -L "https://github.com/modelcontextprotocol/registry/releases/download/v1.1.0/mcp-publisher_1.1.0_$(uname -s | tr '[:upper:]' '[:lower:]')_$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/').tar.gz" | tar xz mcp-publisher
- name: Login to MCP Registry (DNS auth)
run: |
# Extract private key using official MCP publisher method
echo "${{ secrets.MCP_PRIVATE_KEY }}" > temp_key.pem
PRIVATE_KEY_HEX=$(openssl pkey -in temp_key.pem -noout -text | grep -A3 "priv:" | tail -n +2 | tr -d ' :\n')
./mcp-publisher login dns --domain pipeboard.co --private-key "$PRIVATE_KEY_HEX"
rm -f temp_key.pem
- name: Publish to MCP Registry
run: ./mcp-publisher publish
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/ads_library.py:
--------------------------------------------------------------------------------
```python
"""Adds Library-related functionality for Meta Ads API."""
import json
import os
from typing import Optional, List, Dict, Any
from .api import meta_api_tool, make_api_request
from .server import mcp_server
# Only register the search_ads_archive function if the environment variable is NOT set
DISABLE_ADS_LIBRARY = bool(os.environ.get("META_ADS_DISABLE_ADS_LIBRARY", ""))
if not DISABLE_ADS_LIBRARY:
@mcp_server.tool()
@meta_api_tool
async def search_ads_archive(
search_terms: str,
ad_reached_countries: List[str],
access_token: Optional[str] = None,
ad_type: str = "ALL",
limit: int = 25, # Default limit, adjust as needed
fields: str = "ad_creation_time,ad_creative_body,ad_creative_link_caption,ad_creative_link_description,ad_creative_link_title,ad_delivery_start_time,ad_delivery_stop_time,ad_snapshot_url,currency,demographic_distribution,funding_entity,impressions,page_id,page_name,publisher_platform,region_distribution,spend"
) -> str:
"""
Search the Facebook Ads Library archive.
Args:
search_terms: The search query for ads.
ad_reached_countries: List of country codes (e.g., ["US", "GB"]).
access_token: Meta API access token (optional - will use cached token if not provided).
ad_type: Type of ads to search for (e.g., POLITICAL_AND_ISSUE_ADS, HOUSING_ADS, ALL).
limit: Maximum number of ads to return.
fields: Comma-separated string of fields to retrieve for each ad.
Example Usage via curl equivalent:
curl -G \\
-d "search_terms='california'" \\
-d "ad_type=POLITICAL_AND_ISSUE_ADS" \\
-d "ad_reached_countries=['US']" \\
-d "fields=ad_snapshot_url,spend" \\
-d "access_token=<ACCESS_TOKEN>" \\
"https://graph.facebook.com/<API_VERSION>/ads_archive"
"""
if not access_token:
# Attempt to get token implicitly if not provided - meta_api_tool handles this
pass
if not search_terms:
return json.dumps({"error": "search_terms parameter is required"}, indent=2)
if not ad_reached_countries:
return json.dumps({"error": "ad_reached_countries parameter is required"}, indent=2)
endpoint = "ads_archive"
params = {
"search_terms": search_terms,
"ad_type": ad_type,
"ad_reached_countries": json.dumps(ad_reached_countries), # API expects a JSON array string
"limit": limit,
"fields": fields,
}
try:
data = await make_api_request(endpoint, access_token, params, method="GET")
return json.dumps(data, indent=2)
except Exception as e:
error_msg = str(e)
# Consider logging the full error for debugging
# print(f"Error calling Ads Library API: {error_msg}")
return json.dumps({
"error": "Failed to search ads archive",
"details": error_msg,
"params_sent": {k: v for k, v in params.items() if k != 'access_token'} # Avoid logging token
}, indent=2)
```
--------------------------------------------------------------------------------
/CUSTOM_META_APP.md:
--------------------------------------------------------------------------------
```markdown
# Using a Custom Meta Developer App
This guide explains how to use Meta Ads MCP with your own Meta Developer App. Note that this is an alternative method - we recommend using [Pipeboard authentication](README.md) for a simpler setup.
## Create a Meta Developer App
Before using direct Meta authentication, you'll need to set up a Meta Developer App:
1. Go to [Meta for Developers](https://developers.facebook.com/) and create a new app
2. Choose the "Business" app type
3. In your app settings, add the "Marketing API" product
4. Configure your app's OAuth redirect URI to include `http://localhost:8888/callback`
5. Note your App ID (Client ID) for use with the MCP
## Installation & Usage
When using your own Meta app, you'll need to provide the App ID:
```bash
# Using uvx
uvx meta-ads-mcp --app-id YOUR_META_ADS_APP_ID
# Using pip installation
python -m meta_ads_mcp --app-id YOUR_META_ADS_APP_ID
```
## Configuration
### Cursor or Claude Desktop Integration
Add this to your `claude_desktop_config.json` or `~/.cursor/mcp.json`:
```json
"mcpServers": {
"meta-ads": {
"command": "uvx",
"args": ["meta-ads-mcp", "--app-id", "YOUR_META_ADS_APP_ID"]
}
}
```
## Authentication Flow
When using direct Meta OAuth, the MCP uses Meta's OAuth 2.0 authentication flow:
1. Starts a local callback server on your machine
2. Opens a browser window to authenticate with Meta
3. Asks you to authorize the app
4. Redirects back to the local server to extract and store the token securely
## Environment Variables
You can use these environment variables instead of command-line arguments:
```bash
# Your Meta App ID
export META_APP_ID=your_app_id
uvx meta-ads-mcp
# Or provide a direct access token (bypasses local cache)
export META_ACCESS_TOKEN=your_access_token
uvx meta-ads-mcp
```
## Testing
### CLI Testing
Run the test script to verify authentication:
```bash
# Basic test
python test_meta_ads_auth.py --app-id YOUR_APP_ID
# Force new login
python test_meta_ads_auth.py --app-id YOUR_APP_ID --force-login
```
### LLM Interface Testing
When using direct Meta authentication:
1. Test authentication by calling the `mcp_meta_ads_get_login_link` tool
2. Verify account access by calling `mcp_meta_ads_get_ad_accounts`
3. Check specific account details with `mcp_meta_ads_get_account_info`
## Troubleshooting
### Authentication Issues
1. App ID Issues
- If you encounter errors like `(#200) Provide valid app ID`, verify your App ID is correct
- Make sure you've completed the app setup steps above
- Check that your app has the Marketing API product added
2. OAuth Flow
- Run with `--force-login` to get a fresh token: `uvx meta-ads-mcp --login --app-id YOUR_APP_ID --force-login`
- Make sure the terminal has permissions to open a browser window
- Check that the callback server can run on port 8888
3. Direct Token Usage
- If you have a valid access token, you can bypass the OAuth flow:
- `export META_ACCESS_TOKEN=your_access_token`
- This will ignore the local token cache
### API Errors
If you receive errors from the Meta API:
1. Verify your app has the Marketing API product added
2. Ensure the user has appropriate permissions on the ad accounts
3. Check if there are rate limits or other restrictions on your app
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/accounts.py:
--------------------------------------------------------------------------------
```python
"""Account-related functionality for Meta Ads API."""
import json
from typing import Optional, Dict, Any
from .api import meta_api_tool, make_api_request
from .server import mcp_server
@mcp_server.tool()
@meta_api_tool
async def get_ad_accounts(access_token: Optional[str] = None, user_id: str = "me", limit: int = 200) -> str:
"""
Get ad accounts accessible by a user.
Args:
access_token: Meta API access token (optional - will use cached token if not provided)
user_id: Meta user ID or "me" for the current user
limit: Maximum number of accounts to return (default: 200)
"""
endpoint = f"{user_id}/adaccounts"
params = {
"fields": "id,name,account_id,account_status,amount_spent,balance,currency,age,business_city,business_country_code",
"limit": limit
}
data = await make_api_request(endpoint, access_token, params)
return json.dumps(data, indent=2)
@mcp_server.tool()
@meta_api_tool
async def get_account_info(account_id: str, access_token: Optional[str] = None) -> str:
"""
Get detailed information about a specific ad account.
Args:
account_id: Meta Ads account ID (format: act_XXXXXXXXX)
access_token: Meta API access token (optional - will use cached token if not provided)
"""
if not account_id:
return {
"error": {
"message": "Account ID is required",
"details": "Please specify an account_id parameter",
"example": "Use account_id='act_123456789' or account_id='123456789'"
}
}
# Ensure account_id has the 'act_' prefix for API compatibility
if not account_id.startswith("act_"):
account_id = f"act_{account_id}"
# Try to get the account info directly first
endpoint = f"{account_id}"
params = {
"fields": "id,name,account_id,account_status,amount_spent,balance,currency,age,business_city,business_country_code,timezone_name"
}
data = await make_api_request(endpoint, access_token, params)
# Check if the API request returned an error
if "error" in data:
# If access was denied, provide helpful error message with accessible accounts
if "access" in str(data.get("error", {})).lower() or "permission" in str(data.get("error", {})).lower():
# Get list of accessible accounts for helpful error message
accessible_endpoint = "me/adaccounts"
accessible_params = {
"fields": "id,name,account_id,account_status,amount_spent,balance,currency,age,business_city,business_country_code",
"limit": 50
}
accessible_accounts_data = await make_api_request(accessible_endpoint, access_token, accessible_params)
if "data" in accessible_accounts_data:
accessible_accounts = [
{"id": acc["id"], "name": acc["name"]}
for acc in accessible_accounts_data["data"][:10] # Show first 10
]
return {
"error": {
"message": f"Account {account_id} is not accessible to your user account",
"details": "This account either doesn't exist or you don't have permission to access it",
"accessible_accounts": accessible_accounts,
"total_accessible_accounts": len(accessible_accounts_data["data"]),
"suggestion": "Try using one of the accessible account IDs listed above"
}
}
# Return the original error for non-permission related issues
return data
# Add DSA requirement detection
if "business_country_code" in data:
european_countries = ["DE", "FR", "IT", "ES", "NL", "BE", "AT", "IE", "DK", "SE", "FI", "NO"]
if data["business_country_code"] in european_countries:
data["dsa_required"] = True
data["dsa_compliance_note"] = "This account is subject to European DSA (Digital Services Act) requirements"
else:
data["dsa_required"] = False
data["dsa_compliance_note"] = "This account is not subject to European DSA requirements"
return data
```
--------------------------------------------------------------------------------
/tests/test_upload_ad_image.py:
--------------------------------------------------------------------------------
```python
import json
from unittest.mock import AsyncMock, patch
import pytest
from meta_ads_mcp.core.ads import upload_ad_image
@pytest.mark.asyncio
async def test_upload_ad_image_normalizes_images_dict():
mock_response = {
"images": {
"abc123": {
"hash": "abc123",
"url": "https://example.com/image.jpg",
"width": 1200,
"height": 628,
"name": "image.jpg",
"status": 1,
}
}
}
with patch("meta_ads_mcp.core.ads.make_api_request", new_callable=AsyncMock) as mock_api:
mock_api.return_value = mock_response
# Use a data URL input to exercise that branch
file_data_url = "data:image/png;base64,QUJDREVGRw==" # 'ABCDEFG' base64
result_json = await upload_ad_image(
access_token="test",
account_id="act_123",
file=file_data_url,
name="my-upload.png",
)
result = json.loads(result_json)
assert result.get("success") is True
assert result.get("account_id") == "act_123"
assert result.get("name") == "my-upload.png"
assert result.get("image_hash") == "abc123"
assert result.get("images_count") == 1
assert isinstance(result.get("images"), list) and result["images"][0]["hash"] == "abc123"
@pytest.mark.asyncio
async def test_upload_ad_image_error_structure_is_surfaced():
mock_response = {"error": {"message": "Something went wrong", "code": 400}}
with patch("meta_ads_mcp.core.ads.make_api_request", new_callable=AsyncMock) as mock_api:
mock_api.return_value = mock_response
result_json = await upload_ad_image(
access_token="test",
account_id="act_123",
file="data:image/png;base64,QUJD",
)
# Error responses from MCP functions may be wrapped multiple times under a data field
payload = result_json
for _ in range(5):
# If it's a JSON string, parse it
if isinstance(payload, str):
try:
payload = json.loads(payload)
continue
except Exception:
break
# If it's a dict containing a JSON string in data, unwrap once
if isinstance(payload, dict) and "data" in payload:
payload = payload["data"]
continue
break
error_payload = payload if isinstance(payload, dict) else json.loads(payload)
assert "error" in error_payload
assert error_payload["error"] == "Failed to upload image"
assert isinstance(error_payload.get("details"), dict)
assert error_payload.get("account_id") == "act_123"
@pytest.mark.asyncio
async def test_upload_ad_image_fallback_wraps_raw_response():
mock_response = {"unexpected": "shape"}
with patch("meta_ads_mcp.core.ads.make_api_request", new_callable=AsyncMock) as mock_api:
mock_api.return_value = mock_response
result_json = await upload_ad_image(
access_token="test",
account_id="act_123",
file="data:image/png;base64,QUJD",
)
result = json.loads(result_json)
assert result.get("success") is True
assert result.get("raw_response") == mock_response
@pytest.mark.asyncio
async def test_upload_ad_image_from_url_infers_name_and_prefixes_account_id():
mock_response = {
"images": {
"hash999": {
"url": "https://example.com/img.jpg",
"width": 800,
"height": 600,
# omit nested hash intentionally to test normalization fallback
}
}
}
with patch("meta_ads_mcp.core.ads.try_multiple_download_methods", new_callable=AsyncMock) as mock_dl, \
patch("meta_ads_mcp.core.ads.make_api_request", new_callable=AsyncMock) as mock_api:
mock_dl.return_value = b"\xff\xd8\xff" # minimal JPEG header bytes
mock_api.return_value = mock_response
# Provide raw account id (without act_) and ensure it is prefixed in the result
result_json = await upload_ad_image(
access_token="test",
account_id="701351919139047",
image_url="https://cdn.example.com/path/photo.jpg?x=1",
)
result = json.loads(result_json)
assert result.get("success") is True
assert result.get("account_id") == "act_701351919139047"
# Name should be inferred from URL
assert result.get("name") == "photo.jpg"
# Primary hash should be derived from key when nested hash missing
assert result.get("image_hash") == "hash999"
assert result.get("images_count") == 1
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/insights.py:
--------------------------------------------------------------------------------
```python
"""Insights and Reporting functionality for Meta Ads API."""
import json
from typing import Optional, Union, Dict
from .api import meta_api_tool, make_api_request
from .utils import download_image, try_multiple_download_methods, ad_creative_images, create_resource_from_image
from .server import mcp_server
import base64
import datetime
@mcp_server.tool()
@meta_api_tool
async def get_insights(object_id: str, access_token: Optional[str] = None,
time_range: Union[str, Dict[str, str]] = "maximum", breakdown: str = "",
level: str = "ad", limit: int = 25, after: str = "") -> str:
"""
Get performance insights for a campaign, ad set, ad or account.
Args:
object_id: ID of the campaign, ad set, ad or account
access_token: Meta API access token (optional - will use cached token if not provided)
time_range: Either a preset time range string or a dictionary with "since" and "until" dates in YYYY-MM-DD format
Preset options: today, yesterday, this_month, last_month, this_quarter, maximum, data_maximum,
last_3d, last_7d, last_14d, last_28d, last_30d, last_90d, last_week_mon_sun,
last_week_sun_sat, last_quarter, last_year, this_week_mon_today, this_week_sun_today, this_year
Dictionary example: {"since":"2023-01-01","until":"2023-01-31"}
breakdown: Optional breakdown dimension. Valid values include:
Demographic: age, gender, country, region, dma
Platform/Device: device_platform, platform_position, publisher_platform, impression_device
Creative Assets: ad_format_asset, body_asset, call_to_action_asset, description_asset,
image_asset, link_url_asset, title_asset, video_asset, media_asset_url,
media_creator, media_destination_url, media_format, media_origin_url,
media_text_content, media_type, creative_relaxation_asset_type,
flexible_format_asset_type, gen_ai_asset_type
Campaign/Ad Attributes: breakdown_ad_objective, breakdown_reporting_ad_id, app_id, product_id
Conversion Tracking: coarse_conversion_value, conversion_destination, standard_event_content_type,
signal_source_bucket, is_conversion_id_modeled, fidelity_type, redownload
Time-based: hourly_stats_aggregated_by_advertiser_time_zone,
hourly_stats_aggregated_by_audience_time_zone, frequency_value
Extensions/Landing: ad_extension_domain, ad_extension_url, landing_destination,
mdsa_landing_destination
Attribution: sot_attribution_model_type, sot_attribution_window, sot_channel,
sot_event_type, sot_source
Mobile/SKAN: skan_campaign_id, skan_conversion_id, skan_version, postback_sequence_index
CRM/Business: crm_advertiser_l12_territory_ids, crm_advertiser_subvertical_id,
crm_advertiser_vertical_id, crm_ult_advertiser_id, user_persona_id, user_persona_name
Advanced: hsid, is_auto_advance, is_rendered_as_delayed_skip_ad, mmm, place_page_id,
marketing_messages_btn_name, impression_view_time_advertiser_hour_v2, comscore_market,
comscore_market_code
level: Level of aggregation (ad, adset, campaign, account)
limit: Maximum number of results to return per page (default: 25, Meta API allows much higher values)
after: Pagination cursor to get the next set of results. Use the 'after' cursor from previous response's paging.next field.
"""
if not object_id:
return json.dumps({"error": "No object ID provided"}, indent=2)
endpoint = f"{object_id}/insights"
params = {
"fields": "account_id,account_name,campaign_id,campaign_name,adset_id,adset_name,ad_id,ad_name,impressions,clicks,spend,cpc,cpm,ctr,reach,frequency,actions,action_values,conversions,unique_clicks,cost_per_action_type",
"level": level,
"limit": limit
}
# Handle time range based on type
if isinstance(time_range, dict):
# Use custom date range with since/until parameters
if "since" in time_range and "until" in time_range:
params["time_range"] = json.dumps(time_range)
else:
return json.dumps({"error": "Custom time_range must contain both 'since' and 'until' keys in YYYY-MM-DD format"}, indent=2)
else:
# Use preset date range
params["date_preset"] = time_range
if breakdown:
params["breakdowns"] = breakdown
if after:
params["after"] = after
data = await make_api_request(endpoint, access_token, params)
return json.dumps(data, indent=2)
```
--------------------------------------------------------------------------------
/RELEASE.md:
--------------------------------------------------------------------------------
```markdown
# Release Process
This repository uses GitHub Actions to automatically publish releases to PyPI. Here's the optimized release process:
## 🚀 Quick Release (Recommended)
### Prerequisites
- ✅ **Trusted Publishing Configured**: Repository uses PyPI trusted publishing with OIDC tokens
- ✅ **GitHub CLI installed**: `gh` command available for streamlined releases
- ✅ **Clean working directory**: No uncommitted changes
### Optimal Release Process
1. **Update version in three files** (use consistent versioning):
```bash
# Update pyproject.toml
sed -i '' 's/version = "0.7.7"/version = "0.7.8"/' pyproject.toml
# Update __init__.py
sed -i '' 's/__version__ = "0.7.7"/__version__ = "0.7.8"/' meta_ads_mcp/__init__.py
# Update server.json (both top-level and package versions)
sed -i '' 's/"version": "0.7.7"/"version": "0.7.8"/g' server.json
```
Or manually edit:
- `pyproject.toml`: `version = "0.7.8"`
- `meta_ads_mcp/__init__.py`: `__version__ = "0.7.8"`
- `server.json`: set `"version": "0.7.8"` and package `"version": "0.7.8"`
2. **Commit and push version changes**:
```bash
git add pyproject.toml meta_ads_mcp/__init__.py server.json
git commit -m "Bump version to 0.7.8"
git push origin main
```
3. **Create GitHub release** (triggers automatic PyPI publishing):
```bash
# Use bash wrapper if gh has issues in Cursor
bash -c "gh release create 0.7.8 --title '0.7.8' --generate-notes"
```
4. **Verify release** (optional):
```bash
# Check GitHub release
curl -s "https://api.github.com/repos/pipeboard-co/meta-ads-mcp/releases/latest" | grep -E '"tag_name"|"name"'
# Check PyPI availability (wait 2-3 minutes)
curl -s "https://pypi.org/pypi/meta-ads-mcp/json" | grep -E '"version"|"0.7.8"'
```
## 📋 Detailed Release Process
### Version Management Best Practices
- **Semantic Versioning**: Follow `MAJOR.MINOR.PATCH` (e.g., 0.7.8)
- **Synchronized Files**: Always update BOTH version files
- **Commit Convention**: Use `"Bump version to X.Y.Z"` format
- **Release Tag**: GitHub release tag matches version (no "v" prefix)
### Pre-Release Checklist
```bash
# 1. Ensure clean working directory
git status
# 2. Run tests locally (optional but recommended)
uv run python -m pytest tests/ -v
# 3. Check current version
grep -E 'version =|__version__|"version":' pyproject.toml meta_ads_mcp/__init__.py server.json
```
### Release Commands (One-liner)
```bash
# Complete release in one sequence
VERSION="0.7.8" && \
sed -i '' "s/version = \"0.7.7\"/version = \"$VERSION\"/" pyproject.toml && \
sed -i '' "s/__version__ = \"0.7.7\"/__version__ = \"$VERSION\"/" meta_ads_mcp/__init__.py && \
sed -i '' "s/\"version\": \"0.7.7\"/\"version\": \"$VERSION\"/g" server.json && \
git add pyproject.toml meta_ads_mcp/__init__.py server.json && \
git commit -m "Bump version to $VERSION" && \
git push origin main && \
bash -c "gh release create $VERSION --title '$VERSION' --generate-notes"
```
## 🔄 Workflows
### `publish.yml` (Automatic)
- **Trigger**: GitHub release creation
- **Purpose**: Build and publish to PyPI
- **Security**: OIDC tokens (no API keys)
- **Status**: ✅ Fully automated
### `test.yml` (Validation)
- **Trigger**: Push to main/master
- **Purpose**: Package structure validation
- **Matrix**: Python 3.10, 3.11, 3.12
- **Note**: Build tests only, not pytest
## 🛠️ Troubleshooting
### Common Issues
1. **gh command issues in Cursor**:
```bash
# Use bash wrapper
bash -c "gh release create 0.7.8 --title '0.7.8' --generate-notes"
```
2. **Version mismatch**:
```bash
# Verify all three files have the same version
grep -E 'version =|__version__|"version":' pyproject.toml meta_ads_mcp/__init__.py server.json
```
3. **PyPI not updated**:
```bash
# Check if package is available (wait 2-3 minutes)
curl -s "https://pypi.org/pypi/meta-ads-mcp/json" | grep '"version"'
```
### Manual Deployment (Fallback)
```bash
# Install build tools
pip install build twine
# Build package
python -m build
# Upload to PyPI (requires API token)
python -m twine upload dist/*
```
## 📊 Release Verification
### GitHub Release
- ✅ Release created with correct tag
- ✅ Auto-generated notes from commits
- ✅ Actions tab shows successful workflow
### PyPI Package
- ✅ Package available for installation
- ✅ Correct version displayed
- ✅ All dependencies listed
### Installation Test
```bash
# Test new version installation
pip install meta-ads-mcp==0.7.8
# or
uvx [email protected]
```
## 🔒 Security Notes
- **Trusted Publishing**: Uses GitHub OIDC tokens (no API keys needed)
- **Isolated Builds**: All builds run in GitHub-hosted runners
- **Access Control**: Only maintainers can create releases
- **Audit Trail**: All releases tracked in GitHub Actions
## 📈 Release Metrics
Track successful releases:
- **GitHub Releases**: https://github.com/pipeboard-co/meta-ads-mcp/releases
- **PyPI Package**: https://pypi.org/project/meta-ads-mcp/
- **Actions History**: https://github.com/pipeboard-co/meta-ads-mcp/actions
```
--------------------------------------------------------------------------------
/tests/test_create_ad_creative_simple.py:
--------------------------------------------------------------------------------
```python
"""Test that create_ad_creative handles simple creatives correctly."""
import pytest
import json
from unittest.mock import AsyncMock, patch
from meta_ads_mcp.core.ads import create_ad_creative
@pytest.mark.asyncio
async def test_simple_creative_uses_object_story_spec():
"""Test that singular headline/description uses object_story_spec, not asset_feed_spec."""
# Mock the make_api_request function
with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api, \
patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover:
# Mock page discovery
mock_discover.return_value = {
"success": True,
"page_id": "123456789",
"page_name": "Test Page"
}
# Mock creative creation response
mock_api.side_effect = [
# First call: Create creative
{"id": "creative_123"},
# Second call: Get creative details
{
"id": "creative_123",
"name": "Test Creative",
"status": "ACTIVE"
}
]
# Call create_ad_creative with singular headline and description
result = await create_ad_creative(
account_id="act_701351919139047",
image_hash="test_hash_123",
name="Math Problem - Hormozi",
link_url="https://adrocketx.ai/",
message="If you're spending 4+ hours per campaign...",
headline="Stop paying yourself $12.50/hour",
description="AI builds campaigns in 3min. 156% higher conversions. Free beta.",
call_to_action_type="LEARN_MORE",
access_token="test_token"
)
# Check that make_api_request was called
assert mock_api.call_count == 2
# Get the creative_data that was sent to the API
create_call_args = mock_api.call_args_list[0]
endpoint = create_call_args[0][0]
creative_data = create_call_args[0][2]
print("Creative data sent to API:")
print(json.dumps(creative_data, indent=2))
# Verify it uses object_story_spec, NOT asset_feed_spec
assert "object_story_spec" in creative_data, "Should use object_story_spec for simple creatives"
assert "asset_feed_spec" not in creative_data, "Should NOT use asset_feed_spec for simple creatives"
# Verify object_story_spec structure
assert "link_data" in creative_data["object_story_spec"]
link_data = creative_data["object_story_spec"]["link_data"]
# Verify simple creative fields are in link_data
assert link_data["image_hash"] == "test_hash_123"
assert link_data["link"] == "https://adrocketx.ai/"
assert link_data["message"] == "If you're spending 4+ hours per campaign..."
# The issue: headline and description should be in link_data for simple creatives
# Not in asset_feed_spec
print("\nlink_data structure:")
print(json.dumps(link_data, indent=2))
@pytest.mark.asyncio
async def test_dynamic_creative_uses_asset_feed_spec():
"""Test that plural headlines/descriptions uses asset_feed_spec."""
with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api, \
patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover:
# Mock page discovery
mock_discover.return_value = {
"success": True,
"page_id": "123456789",
"page_name": "Test Page"
}
# Mock creative creation response
mock_api.side_effect = [
{"id": "creative_456"},
{"id": "creative_456", "name": "Dynamic Creative", "status": "ACTIVE"}
]
# Call with PLURAL headlines and descriptions (dynamic creative)
result = await create_ad_creative(
account_id="act_701351919139047",
image_hash="test_hash_456",
name="Dynamic Creative Test",
link_url="https://example.com/",
message="Test message",
headlines=["Headline 1", "Headline 2"],
descriptions=["Description 1", "Description 2"],
access_token="test_token"
)
# Get the creative_data that was sent to the API
create_call_args = mock_api.call_args_list[0]
creative_data = create_call_args[0][2]
print("\nDynamic creative data sent to API:")
print(json.dumps(creative_data, indent=2))
# Verify it uses asset_feed_spec for dynamic creatives
assert "asset_feed_spec" in creative_data, "Should use asset_feed_spec for dynamic creatives"
# Verify asset_feed_spec structure
asset_feed_spec = creative_data["asset_feed_spec"]
assert "headlines" in asset_feed_spec
assert len(asset_feed_spec["headlines"]) == 2
assert "descriptions" in asset_feed_spec
assert len(asset_feed_spec["descriptions"]) == 2
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/reports.py:
--------------------------------------------------------------------------------
```python
"""Report generation functionality for Meta Ads API."""
import json
import os
from typing import Optional, Dict, Any, List, Union
from .api import meta_api_tool
from .server import mcp_server
# Only register the generate_report function if the environment variable is set
ENABLE_REPORT_GENERATION = bool(os.environ.get("META_ADS_ENABLE_REPORTS", ""))
if ENABLE_REPORT_GENERATION:
@mcp_server.tool()
async def generate_report(
account_id: str,
access_token: Optional[str] = None,
report_type: str = "account",
time_range: str = "last_30d",
campaign_ids: Optional[List[str]] = None,
export_format: str = "pdf",
report_name: Optional[str] = None,
include_sections: Optional[List[str]] = None,
breakdowns: Optional[List[str]] = None,
comparison_period: Optional[str] = None
) -> str:
"""
Generate comprehensive Meta Ads performance reports.
**This is a premium feature available with Pipeboard Pro.**
Args:
account_id: Meta Ads account ID (format: act_XXXXXXXXX)
access_token: Meta API access token (optional - will use cached token if not provided)
report_type: Type of report to generate (account, campaign, comparison)
time_range: Time period for the report (e.g., 'last_30d', 'last_7d', 'this_month')
campaign_ids: Specific campaign IDs (required for campaign/comparison reports)
export_format: Output format for the report (pdf, json, html)
report_name: Custom name for the report (auto-generated if not provided)
include_sections: Specific sections to include in the report
breakdowns: Audience breakdown dimensions (age, gender, country, etc.)
comparison_period: Time period for comparison analysis
"""
# Validate required parameters
if not account_id:
return json.dumps({
"error": "invalid_parameters",
"message": "Account ID is required",
"details": {
"required_parameter": "account_id",
"format": "act_XXXXXXXXX"
}
}, indent=2)
# For campaign and comparison reports, campaign_ids are required
if report_type in ["campaign", "comparison"] and not campaign_ids:
return json.dumps({
"error": "invalid_parameters",
"message": f"Campaign IDs are required for {report_type} reports",
"details": {
"required_parameter": "campaign_ids",
"format": "Array of campaign ID strings"
}
}, indent=2)
# Return premium feature upgrade message
return json.dumps({
"error": "premium_feature_required",
"message": "Professional report generation is a premium feature",
"details": {
"feature": "Automated PDF Report Generation",
"description": "Create professional client-ready reports with performance insights, recommendations, and white-label branding",
"benefits": [
"Executive summary with key metrics",
"Performance breakdowns and trends",
"Audience insights and recommendations",
"Professional PDF formatting",
"White-label branding options",
"Campaign comparison analysis",
"Creative performance insights",
"Automated scheduling options"
],
"upgrade_url": "https://pipeboard.co/upgrade",
"contact_email": "[email protected]",
"early_access": "Contact us for early access and special pricing"
},
"request_parameters": {
"account_id": account_id,
"report_type": report_type,
"time_range": time_range,
"export_format": export_format,
"campaign_ids": campaign_ids or [],
"include_sections": include_sections or [],
"breakdowns": breakdowns or []
},
"preview": {
"available_data": {
"account_name": f"Account {account_id}",
"campaigns_count": len(campaign_ids) if campaign_ids else "All campaigns",
"time_range": time_range,
"estimated_report_pages": 8 if report_type == "account" else 6,
"report_format": export_format.upper()
},
"sample_metrics": {
"total_spend": "$12,450",
"total_impressions": "2.3M",
"total_clicks": "45.2K",
"average_cpc": "$0.85",
"average_cpm": "$15.20",
"click_through_rate": "1.96%",
"roas": "4.2x"
},
"available_sections": [
"executive_summary",
"performance_overview",
"campaign_breakdown",
"audience_insights",
"creative_performance",
"recommendations",
"appendix"
],
"supported_breakdowns": [
"age",
"gender",
"country",
"region",
"placement",
"device_platform",
"publisher_platform"
]
}
}, indent=2)
```
--------------------------------------------------------------------------------
/tests/README_REGRESSION_TESTS.md:
--------------------------------------------------------------------------------
```markdown
# Duplication Module Regression Tests
This document describes the comprehensive regression test suite for the Meta Ads duplication module (`meta_ads_mcp/core/duplication.py`).
## Test Coverage Overview
The regression test suite (`test_duplication_regression.py`) contains **23 comprehensive tests** organized into 7 test classes, providing extensive coverage to prevent future regressions.
### 🎯 Test Classes
#### 1. `TestDuplicationFeatureToggle` (4 tests)
- **Purpose**: Ensures the feature toggle mechanism works correctly
- **Coverage**:
- Feature disabled by default
- Feature enabled with environment variable
- Various truthy values enable the feature
- Empty string disables the feature
- **Prevents**: Accidental feature activation, broken environment variable handling
#### 2. `TestDuplicationDecorators` (2 tests)
- **Purpose**: Validates that all decorators are applied correctly
- **Coverage**:
- `@meta_api_tool` decorator applied to all functions
- `@mcp_server.tool()` decorator registers functions as MCP tools
- **Prevents**: Functions missing required decorators, broken MCP registration
#### 3. `TestDuplicationAPIContract` (3 tests)
- **Purpose**: Ensures external API calls follow the correct contract
- **Coverage**:
- API endpoint URL construction
- HTTP request headers format
- Request timeout configuration
- **Prevents**: Broken API integration, malformed requests
#### 4. `TestDuplicationErrorHandling` (3 tests)
- **Purpose**: Validates robust error handling across all scenarios
- **Coverage**:
- Missing access token errors
- HTTP status code handling (200, 401, 403, 429, 500)
- Network error handling (timeouts, connection failures)
- **Prevents**: Unhandled errors, poor error messages, broken error paths
#### 5. `TestDuplicationParameterHandling` (3 tests)
- **Purpose**: Tests parameter processing and forwarding
- **Coverage**:
- None values filtered from options
- Parameter forwarding accuracy
- Estimated components calculation
- **Prevents**: Malformed API requests, parameter corruption
#### 6. `TestDuplicationIntegration` (2 tests)
- **Purpose**: End-to-end functionality testing
- **Coverage**:
- Successful duplication flow
- Premium feature upgrade flow
- **Prevents**: Broken end-to-end flows, integration failures
#### 7. `TestDuplicationTokenHandling` (2 tests)
- **Purpose**: Access token management and injection
- **Coverage**:
- Explicit token handling
- Token parameter override behavior
- **Prevents**: Authentication bypasses, token handling bugs
#### 8. `TestDuplicationRegressionEdgeCases` (4 tests)
- **Purpose**: Edge cases and unusual scenarios
- **Coverage**:
- Empty string parameters
- Unicode parameter handling
- Large parameter values
- Module reload safety
- **Prevents**: Edge case failures, data corruption, memory leaks
## 🚀 Key Features Tested
### Authentication & Security
- ✅ Access token validation and injection
- ✅ Authentication error handling
- ✅ App ID validation
- ✅ Secure token forwarding
### API Integration
- ✅ HTTP client configuration
- ✅ Request/response handling
- ✅ Error status code processing
- ✅ Network failure resilience
### Feature Management
- ✅ Environment-based feature toggle
- ✅ Dynamic module loading
- ✅ MCP tool registration
- ✅ Decorator chain validation
### Data Processing
- ✅ Parameter validation and filtering
- ✅ Unicode and special character handling
- ✅ Large value processing
- ✅ JSON serialization/deserialization
### Error Resilience
- ✅ Network timeouts and failures
- ✅ Malformed responses
- ✅ Authentication failures
- ✅ Rate limiting scenarios
## 🛡️ Regression Prevention
These tests specifically prevent the following categories of regressions:
### **Configuration Regressions**
- Feature accidentally enabled/disabled
- Environment variable handling changes
- Default configuration drift
### **Integration Regressions**
- API endpoint URL changes
- Request format modifications
- Authentication system changes
### **Error Handling Regressions**
- Silent error failures
- Poor error message quality
- Unhandled exception scenarios
### **Performance Regressions**
- Memory leaks in module reloading
- Inefficient parameter processing
- Network timeout misconfigurations
### **Security Regressions**
- Token handling vulnerabilities
- Authentication bypass bugs
- Parameter injection attacks
## 🔧 Running the Tests
```bash
# Run all regression tests
python -m pytest tests/test_duplication_regression.py -v
# Run specific test class
python -m pytest tests/test_duplication_regression.py::TestDuplicationFeatureToggle -v
# Run with coverage
python -m pytest tests/test_duplication_regression.py --cov=meta_ads_mcp.core.duplication
# Run with detailed output
python -m pytest tests/test_duplication_regression.py -vvv --tb=long
```
## 📊 Test Results
When all tests pass, you should see:
```
====================== 23 passed, 5 warnings in 0.54s ======================
```
The warnings are from mock objects and don't affect functionality.
## 🔍 Test Design Principles
1. **Isolation**: Each test is independent and can run standalone
2. **Mocking**: External dependencies are mocked for reliability
3. **Comprehensive**: Cover both happy path and error scenarios
4. **Realistic**: Use realistic data and scenarios
5. **Maintainable**: Clear test names and documentation
## 🚨 Adding New Tests
When adding new functionality to the duplication module:
1. **Add corresponding regression tests**
2. **Test both success and failure scenarios**
3. **Mock external dependencies appropriately**
4. **Use descriptive test names**
5. **Update this documentation**
## 📈 Coverage Goals
- **Line Coverage**: > 95%
- **Branch Coverage**: > 90%
- **Function Coverage**: 100%
- **Error Path Coverage**: > 85%
This comprehensive test suite ensures the duplication module remains stable and reliable across future changes and updates.
```
--------------------------------------------------------------------------------
/tests/test_get_ad_creatives_fix.py:
--------------------------------------------------------------------------------
```python
"""Regression tests for get_ad_creatives function bug fix.
Tests for issue where get_ad_creatives would throw:
'TypeError: 'dict' object is not callable'
This was caused by trying to call ad_creative_images(creative) where
ad_creative_images is a dictionary, not a function.
The fix was to create extract_creative_image_urls() function and use that instead.
"""
import pytest
import json
from unittest.mock import AsyncMock, patch
from meta_ads_mcp.core.ads import get_ad_creatives
from meta_ads_mcp.core.utils import ad_creative_images
@pytest.mark.asyncio
class TestGetAdCreativesBugFix:
"""Regression test cases for the get_ad_creatives function bug fix."""
async def test_get_ad_creatives_regression_fix(self):
"""Regression test: ensure get_ad_creatives works correctly and doesn't throw 'dict' object is not callable."""
# Mock the make_api_request to return sample creative data
sample_creative_data = {
"data": [
{
"id": "123456789",
"name": "Test Creative",
"status": "ACTIVE",
"thumbnail_url": "https://example.com/thumb.jpg",
"image_url": "https://example.com/image.jpg",
"image_hash": "abc123",
"object_story_spec": {
"page_id": "987654321",
"link_data": {
"image_hash": "abc123",
"link": "https://example.com",
"name": "Test Ad"
}
}
}
]
}
with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
mock_api.return_value = sample_creative_data
# This should NOT raise a TypeError anymore
# Previously this would fail with: TypeError: 'dict' object is not callable
result = await get_ad_creatives(access_token="test_token", ad_id="120228922933270272")
# Parse the result
result_data = json.loads(result)
# Verify the structure is correct
assert "data" in result_data
assert len(result_data["data"]) == 1
creative = result_data["data"][0]
assert creative["id"] == "123456789"
assert creative["name"] == "Test Creative"
assert "image_urls_for_viewing" in creative
assert isinstance(creative["image_urls_for_viewing"], list)
async def test_extract_creative_image_urls_function(self):
"""Test the extract_creative_image_urls function works correctly."""
from meta_ads_mcp.core.utils import extract_creative_image_urls
# Test creative with various image URL fields
test_creative = {
"id": "123456789",
"name": "Test Creative",
"status": "ACTIVE",
"thumbnail_url": "https://example.com/thumb.jpg",
"image_url": "https://example.com/image.jpg",
"image_hash": "abc123",
"object_story_spec": {
"page_id": "987654321",
"link_data": {
"image_hash": "abc123",
"link": "https://example.com",
"name": "Test Ad",
"picture": "https://example.com/picture.jpg"
}
}
}
urls = extract_creative_image_urls(test_creative)
# Should extract URLs in order: image_url, picture, thumbnail_url (new priority order)
expected_urls = [
"https://example.com/image.jpg",
"https://example.com/picture.jpg",
"https://example.com/thumb.jpg"
]
assert urls == expected_urls
# Test with empty creative
empty_urls = extract_creative_image_urls({})
assert empty_urls == []
# Test with duplicates (should remove them)
duplicate_creative = {
"image_url": "https://example.com/same.jpg",
"thumbnail_url": "https://example.com/same.jpg", # Same URL
}
unique_urls = extract_creative_image_urls(duplicate_creative)
assert unique_urls == ["https://example.com/same.jpg"]
async def test_get_ad_creatives_no_ad_id(self):
"""Test get_ad_creatives with no ad_id provided."""
result = await get_ad_creatives(access_token="test_token", ad_id=None)
result_data = json.loads(result)
# The @meta_api_tool decorator wraps the result in a data field
assert "data" in result_data
error_data = json.loads(result_data["data"])
assert "error" in error_data
assert error_data["error"] == "No ad ID provided"
async def test_get_ad_creatives_empty_data(self):
"""Test get_ad_creatives when API returns empty data."""
empty_data = {"data": []}
with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
mock_api.return_value = empty_data
result = await get_ad_creatives(access_token="test_token", ad_id="120228922933270272")
result_data = json.loads(result)
assert "data" in result_data
assert len(result_data["data"]) == 0
def test_ad_creative_images_is_dict():
"""Test that ad_creative_images is a dictionary, not a function.
This confirms the original issue: ad_creative_images is a dict for storing image data,
but was being called as a function ad_creative_images(creative), which would fail.
This test ensures we don't accidentally change ad_creative_images to a function
and break its intended purpose as a storage dictionary.
"""
assert isinstance(ad_creative_images, dict)
# This would raise TypeError: 'dict' object is not callable
# This is the original bug - trying to call a dict as a function
with pytest.raises(TypeError, match="'dict' object is not callable"):
ad_creative_images({"test": "data"})
```
--------------------------------------------------------------------------------
/examples/example_http_client.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Example HTTP client for Meta Ads MCP Streamable HTTP transport
This demonstrates how to use the completed HTTP transport implementation
to access Meta Ads tools via HTTP API calls.
Usage:
1. Start the server: python -m meta_ads_mcp --transport streamable-http
2. Run this example: python example_http_client.py
"""
import requests
import json
import os
from typing import Dict, Any, Optional
class MetaAdsMCPClient:
"""Simple HTTP client for Meta Ads MCP server"""
def __init__(self, base_url: str = "http://localhost:8080",
pipeboard_token: Optional[str] = None,
meta_access_token: Optional[str] = None):
"""Initialize the client
Args:
base_url: Base URL of the MCP server
pipeboard_token: Pipeboard API token (recommended)
meta_access_token: Direct Meta access token (fallback)
"""
self.base_url = base_url.rstrip('/')
self.endpoint = f"{self.base_url}/mcp/"
self.session_id = 1
# Setup authentication headers
self.headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
"User-Agent": "MetaAdsMCP-Example-Client/1.0"
}
# Add authentication
if pipeboard_token:
self.headers["Authorization"] = f"Bearer {pipeboard_token}"
print(f"✅ Using Pipeboard authentication")
elif meta_access_token:
self.headers["X-META-ACCESS-TOKEN"] = meta_access_token
print(f"✅ Using direct Meta token authentication")
else:
print(f"⚠️ No authentication provided - tools will require auth")
def _make_request(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
"""Make a JSON-RPC request to the server"""
payload = {
"jsonrpc": "2.0",
"method": method,
"id": self.session_id
}
if params:
payload["params"] = params
print(f"\n🔄 Making request: {method}")
print(f" URL: {self.endpoint}")
print(f" Headers: {json.dumps(dict(self.headers), indent=2)}")
print(f" Payload: {json.dumps(payload, indent=2)}")
try:
response = requests.post(
self.endpoint,
headers=self.headers,
json=payload,
timeout=30
)
print(f" Status: {response.status_code} {response.reason}")
print(f" Response Headers: {dict(response.headers)}")
if response.status_code == 200:
result = response.json()
print(f"✅ Request successful")
return result
else:
print(f"❌ Request failed: {response.status_code}")
print(f" Response: {response.text}")
return {"error": {"code": response.status_code, "message": response.text}}
except Exception as e:
print(f"❌ Request exception: {e}")
return {"error": {"code": -1, "message": str(e)}}
finally:
self.session_id += 1
def initialize(self) -> Dict[str, Any]:
"""Initialize MCP session"""
return self._make_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {
"roots": {"listChanged": True},
"sampling": {}
},
"clientInfo": {
"name": "meta-ads-example-client",
"version": "1.0.0"
}
})
def list_tools(self) -> Dict[str, Any]:
"""Get list of available tools"""
return self._make_request("tools/list")
def call_tool(self, tool_name: str, arguments: Dict[str, Any] = None) -> Dict[str, Any]:
"""Call a specific tool"""
params = {"name": tool_name}
if arguments:
params["arguments"] = arguments
return self._make_request("tools/call", params)
def main():
"""Example usage of the Meta Ads MCP HTTP client"""
print("🚀 Meta Ads MCP HTTP Client Example")
print("="*60)
# Check for authentication
pipeboard_token = os.environ.get("PIPEBOARD_API_TOKEN")
meta_token = os.environ.get("META_ACCESS_TOKEN")
if not pipeboard_token and not meta_token:
print("⚠️ No authentication tokens found in environment")
print(" Set PIPEBOARD_API_TOKEN or META_ACCESS_TOKEN for full functionality")
print(" Using test token for demonstration...")
pipeboard_token = "demo_token_12345"
# Create client
client = MetaAdsMCPClient(
pipeboard_token=pipeboard_token,
meta_access_token=meta_token
)
# Test the MCP protocol flow
print("\n🔄 Testing MCP Protocol Flow")
print("="*50)
# 1. Initialize
print("\n" + "="*60)
print("🔍 Step 1: Initialize MCP Session")
print("="*60)
init_result = client.initialize()
if "error" in init_result:
print(f"❌ Initialize failed: {init_result['error']}")
return
print(f"✅ Initialize successful")
print(f" Server info: {init_result['result']['serverInfo']}")
print(f" Protocol version: {init_result['result']['protocolVersion']}")
# 2. List tools
print("\n" + "="*60)
print("🔍 Step 2: List Available Tools")
print("="*60)
tools_result = client.list_tools()
if "error" in tools_result:
print(f"❌ Tools list failed: {tools_result['error']}")
return
tools = tools_result["result"]["tools"]
print(f"✅ Found {len(tools)} tools:")
# Show first few tools
for i, tool in enumerate(tools[:5]):
print(f" {i+1}. {tool['name']}: {tool['description'][:100]}...")
if len(tools) > 5:
print(f" ... and {len(tools) - 5} more tools")
# 3. Test a simple tool call
print("\n" + "="*60)
print("🔍 Step 3: Test Tool Call - get_ad_accounts")
print("="*60)
tool_result = client.call_tool("get_ad_accounts", {"limit": 3})
if "error" in tool_result:
print(f"❌ Tool call failed: {tool_result['error']}")
return
print(f"✅ Tool call successful")
content = tool_result["result"]["content"][0]["text"]
# Parse the response to see if it's authentication or actual data
try:
parsed_content = json.loads(content)
if "error" in parsed_content and "Authentication Required" in parsed_content["error"]["message"]:
print(f"📋 Result: Authentication required (expected with demo token)")
print(f" This confirms the HTTP transport is working!")
print(f" Use a real Pipeboard token for actual data access.")
else:
print(f"📋 Result: {content[:200]}...")
except:
print(f"📋 Raw result: {content[:200]}...")
# Summary
print("\n" + "🎯" * 30)
print("EXAMPLE COMPLETE")
print("🎯" * 30)
print("\n📊 Results:")
print(" Initialize: ✅ SUCCESS")
print(" Tools List: ✅ SUCCESS")
print(" Tool Call: ✅ SUCCESS")
print("\n🎉 Meta Ads MCP HTTP transport is fully functional!")
print("\n💡 Next steps:")
print(" 1. Set PIPEBOARD_API_TOKEN environment variable")
print(" 2. Call any of the 26 available Meta Ads tools")
print(" 3. Build your web application or automation scripts")
if __name__ == "__main__":
main()
```
--------------------------------------------------------------------------------
/tests/test_integration_openai_mcp.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Integration Test for OpenAI MCP functionality with existing Meta Ads tools
This test verifies that:
1. Existing Meta Ads tools still work after adding OpenAI MCP tools
2. New search and fetch tools are properly registered
3. Both old and new tools can coexist without conflicts
Usage:
python tests/test_integration_openai_mcp.py
"""
import sys
import os
import importlib
# Add project root to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
def test_module_imports():
"""Test that all modules can be imported successfully"""
print("🧪 Testing module imports...")
try:
# Test core module import
import meta_ads_mcp.core as core
print("✅ Core module imported successfully")
# Test that mcp_server is available
assert hasattr(core, 'mcp_server'), "mcp_server not found in core module"
print("✅ mcp_server available")
# Test that existing tools are still available
existing_tools = [
'get_ad_accounts', 'get_account_info', 'get_campaigns',
'get_ads', 'get_insights', 'search_ads_archive'
]
for tool in existing_tools:
assert hasattr(core, tool), f"Existing tool {tool} not found"
print(f"✅ All {len(existing_tools)} existing tools available")
# Test that new OpenAI tools are available
openai_tools = ['search', 'fetch']
for tool in openai_tools:
assert hasattr(core, tool), f"OpenAI tool {tool} not found"
print(f"✅ All {len(openai_tools)} OpenAI MCP tools available")
return True
except Exception as e:
print(f"❌ Module import test failed: {e}")
return False
def test_tool_registration():
"""Test that tools are properly registered with the MCP server"""
print("\n🧪 Testing tool registration...")
try:
# Import the server and get registered tools
from meta_ads_mcp.core.server import mcp_server
# Get all registered tools
# Note: FastMCP may not expose tools until runtime, so we'll check the module structure
print("✅ MCP server accessible")
# Test that OpenAI Deep Research module can be imported
from meta_ads_mcp.core import openai_deep_research
print("✅ OpenAI Deep Research module imported")
# Test that the tools are callable
assert callable(openai_deep_research.search), "search tool is not callable"
assert callable(openai_deep_research.fetch), "fetch tool is not callable"
print("✅ OpenAI tools are callable")
return True
except Exception as e:
print(f"❌ Tool registration test failed: {e}")
return False
def test_tool_signatures():
"""Test that tool signatures are correct"""
print("\n🧪 Testing tool signatures...")
try:
from meta_ads_mcp.core.openai_deep_research import search, fetch
import inspect
# Test search tool signature
search_sig = inspect.signature(search)
search_params = list(search_sig.parameters.keys())
# Should have access_token and query parameters
expected_search_params = ['access_token', 'query']
for param in expected_search_params:
assert param in search_params, f"search tool missing parameter: {param}"
print("✅ search tool has correct signature")
# Test fetch tool signature
fetch_sig = inspect.signature(fetch)
fetch_params = list(fetch_sig.parameters.keys())
# Should have id parameter
assert 'id' in fetch_params, "fetch tool missing 'id' parameter"
print("✅ fetch tool has correct signature")
return True
except Exception as e:
print(f"❌ Tool signature test failed: {e}")
return False
def test_existing_functionality_preserved():
"""Test that existing functionality is not broken"""
print("\n🧪 Testing existing functionality preservation...")
try:
# Test that we can still import and access existing tools
from meta_ads_mcp.core.accounts import get_ad_accounts
from meta_ads_mcp.core.campaigns import get_campaigns
from meta_ads_mcp.core.ads import get_ads
from meta_ads_mcp.core.insights import get_insights
# Verify they're still callable
assert callable(get_ad_accounts), "get_ad_accounts is not callable"
assert callable(get_campaigns), "get_campaigns is not callable"
assert callable(get_ads), "get_ads is not callable"
assert callable(get_insights), "get_insights is not callable"
print("✅ All existing tools remain callable")
# Test that existing tool signatures haven't changed
import inspect
accounts_sig = inspect.signature(get_ad_accounts)
accounts_params = list(accounts_sig.parameters.keys())
assert 'access_token' in accounts_params, "get_ad_accounts signature changed"
print("✅ Existing tool signatures preserved")
return True
except Exception as e:
print(f"❌ Existing functionality test failed: {e}")
return False
def test_no_name_conflicts():
"""Test that there are no naming conflicts between old and new tools"""
print("\n🧪 Testing for naming conflicts...")
try:
import meta_ads_mcp.core as core
# Get all attributes from the core module
all_attrs = dir(core)
# Check for expected tools
existing_tools = [
'get_ad_accounts', 'get_campaigns', 'get_ads', 'get_insights',
'search_ads_archive' # This is the existing search function
]
new_tools = ['search', 'fetch'] # These are the new OpenAI tools
# Verify all tools exist
for tool in existing_tools + new_tools:
assert tool in all_attrs, f"Tool {tool} not found in core module"
# Verify search_ads_archive and search are different functions
assert core.search_ads_archive != core.search, "search and search_ads_archive should be different functions"
print("✅ No naming conflicts detected")
print(f" - Existing search tool: search_ads_archive (Meta Ads Library)")
print(f" - New search tool: search (OpenAI MCP Deep Research)")
return True
except Exception as e:
print(f"❌ Naming conflict test failed: {e}")
return False
def main():
"""Run all integration tests"""
print("🚀 OpenAI MCP Integration Tests")
print("="*50)
tests = [
("Module Imports", test_module_imports),
("Tool Registration", test_tool_registration),
("Tool Signatures", test_tool_signatures),
("Existing Functionality", test_existing_functionality_preserved),
("Naming Conflicts", test_no_name_conflicts),
]
results = {}
all_passed = True
for test_name, test_func in tests:
try:
result = test_func()
results[test_name] = result
if not result:
all_passed = False
except Exception as e:
print(f"❌ {test_name} test crashed: {e}")
results[test_name] = False
all_passed = False
# Summary
print("\n🏁 INTEGRATION TEST RESULTS")
print("="*30)
for test_name, result in results.items():
status = "✅ PASSED" if result else "❌ FAILED"
print(f"{test_name}: {status}")
print(f"\n📊 Overall Result: {'✅ ALL TESTS PASSED' if all_passed else '❌ SOME TESTS FAILED'}")
if all_passed:
print("\n🎉 Integration successful!")
print(" • Existing Meta Ads tools: Working")
print(" • New OpenAI MCP tools: Working")
print(" • No conflicts detected")
print(" • Ready for OpenAI ChatGPT Deep Research")
print("\n📋 Next steps:")
print(" 1. Start server inside uv virtual env:")
print(" # Basic HTTP server (default: localhost:8080)")
print(" python -m meta_ads_mcp --transport streamable-http")
print(" ")
print(" # Custom host and port")
print(" python -m meta_ads_mcp --transport streamable-http --host 0.0.0.0 --port 9000")
print(" 2. Run OpenAI tests: python tests/test_openai_mcp_deep_research.py")
else:
print("\n⚠️ Integration issues detected")
print(" Please fix failed tests before proceeding")
return 0 if all_passed else 1
if __name__ == "__main__":
sys.exit(main())
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/callback_server.py:
--------------------------------------------------------------------------------
```python
"""Callback server for Meta Ads API authentication."""
import threading
import socket
import asyncio
import json
import logging
import webbrowser
import os
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs, quote
from typing import Dict, Any, Optional
from .utils import logger
# Global token container for communication between threads
token_container = {"token": None, "expires_in": None, "user_id": None}
# Global variables for server thread and state
callback_server_thread = None
callback_server_lock = threading.Lock()
callback_server_running = False
callback_server_port = None
callback_server_instance = None
server_shutdown_timer = None
# Timeout in seconds before shutting down the callback server
CALLBACK_SERVER_TIMEOUT = 180 # 3 minutes timeout
class CallbackHandler(BaseHTTPRequestHandler):
def do_GET(self):
try:
# Print path for debugging
print(f"Callback server received request: {self.path}")
if self.path.startswith("/callback"):
self._handle_oauth_callback()
elif self.path.startswith("/token"):
self._handle_token()
else:
# If no matching path, return a 404 error
self.send_response(404)
self.end_headers()
except Exception as e:
print(f"Error processing request: {e}")
self.send_response(500)
self.end_headers()
def _handle_oauth_callback(self):
"""Handle OAuth callback after user authorization"""
# Check if we're being redirected from Facebook with an authorization code
parsed_url = urlparse(self.path)
params = parse_qs(parsed_url.query)
# Check for code parameter
code = params.get('code', [None])[0]
state = params.get('state', [None])[0]
error = params.get('error', [None])[0]
# Send 200 OK response with a simple HTML page
self.send_response(200)
self.send_header("Content-type", "text/html")
self.end_headers()
if error:
# User denied access or other error occurred
html = f"""
<html>
<head><title>Authorization Failed</title></head>
<body>
<h1>Authorization Failed</h1>
<p>Error: {error}</p>
<p>The authorization was cancelled or failed. You can close this window.</p>
</body>
</html>
"""
logger.error(f"OAuth authorization failed: {error}")
elif code:
# Success case - we have the authorization code
logger.info(f"Received authorization code: {code[:10]}...")
# Store the authorization code temporarily
# The auth module will exchange this for an access token
token_container.update({
"auth_code": code,
"state": state,
"timestamp": asyncio.get_event_loop().time()
})
html = """
<html>
<head><title>Authorization Successful</title></head>
<body>
<h1>✅ Authorization Successful!</h1>
<p>You have successfully authorized the Meta Ads MCP application.</p>
<p>You can now close this window and return to your application.</p>
<script>
// Try to close the window automatically after 2 seconds
setTimeout(function() {
window.close();
}, 2000);
</script>
</body>
</html>
"""
logger.info("OAuth authorization successful")
else:
# No code or error - something unexpected happened
html = """
<html>
<head><title>Unexpected Response</title></head>
<body>
<h1>Unexpected Response</h1>
<p>No authorization code or error received. Please try again.</p>
</body>
</html>
"""
logger.warning("OAuth callback received without code or error")
self.wfile.write(html.encode())
def _handle_token(self):
"""Handle token endpoint for retrieving stored token data"""
# This endpoint allows other parts of the application to retrieve
# token information from the callback server
self.send_response(200)
self.send_header("Content-type", "application/json")
self.end_headers()
# Return current token container contents
response_data = {
"status": "success",
"data": token_container
}
self.wfile.write(json.dumps(response_data).encode())
# The actual token processing is now handled by the auth module
# that imports this module and accesses token_container
# Silence server logs
def log_message(self, format, *args):
return
def shutdown_callback_server():
"""
Shutdown the callback server if it's running
"""
global callback_server_thread, callback_server_running, callback_server_port, callback_server_instance, server_shutdown_timer
with callback_server_lock:
if not callback_server_running:
print("Callback server is not running")
return
if server_shutdown_timer is not None:
server_shutdown_timer.cancel()
server_shutdown_timer = None
try:
if callback_server_instance:
print("Shutting down callback server...")
callback_server_instance.shutdown()
callback_server_instance.server_close()
print("Callback server shut down successfully")
if callback_server_thread and callback_server_thread.is_alive():
callback_server_thread.join(timeout=5)
if callback_server_thread.is_alive():
print("Warning: Callback server thread did not shut down cleanly")
except Exception as e:
print(f"Error during callback server shutdown: {e}")
finally:
callback_server_running = False
callback_server_thread = None
callback_server_port = None
callback_server_instance = None
def start_callback_server() -> int:
"""
Start the callback server and return the port number it's running on.
Returns:
int: Port number the server is listening on
Raises:
Exception: If the server fails to start
"""
global callback_server_thread, callback_server_running, callback_server_port, callback_server_instance, server_shutdown_timer
# Check if callback server is disabled
if os.environ.get("META_ADS_DISABLE_CALLBACK_SERVER"):
raise Exception("Callback server is disabled via META_ADS_DISABLE_CALLBACK_SERVER environment variable")
with callback_server_lock:
if callback_server_running:
print(f"Callback server already running on port {callback_server_port}")
return callback_server_port
# Find an available port
port = 8080
max_attempts = 10
for attempt in range(max_attempts):
try:
# Test if port is available
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(('localhost', port))
break
except OSError:
port += 1
else:
raise Exception(f"Could not find an available port after {max_attempts} attempts")
callback_server_port = port
# Start the server in a separate thread
callback_server_thread = threading.Thread(target=server_thread, daemon=True)
callback_server_thread.start()
# Wait a moment for the server to start
import time
time.sleep(0.5)
if not callback_server_running:
raise Exception("Failed to start callback server")
# Set up automatic shutdown timer
def auto_shutdown():
print(f"Callback server auto-shutdown after {CALLBACK_SERVER_TIMEOUT} seconds")
shutdown_callback_server()
server_shutdown_timer = threading.Timer(CALLBACK_SERVER_TIMEOUT, auto_shutdown)
server_shutdown_timer.start()
print(f"Callback server started on http://localhost:{port}")
return port
def server_thread():
"""Thread function to run the callback server"""
global callback_server_running, callback_server_instance
try:
callback_server_instance = HTTPServer(('localhost', callback_server_port), CallbackHandler)
callback_server_running = True
print(f"Callback server thread started on port {callback_server_port}")
callback_server_instance.serve_forever()
except Exception as e:
print(f"Callback server error: {e}")
callback_server_running = False
finally:
print("Callback server thread finished")
callback_server_running = False
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/utils.py:
--------------------------------------------------------------------------------
```python
"""Utility functions for Meta Ads API."""
from typing import Optional, Dict, Any, List
import httpx
import io
from PIL import Image as PILImage
import base64
import time
import asyncio
import os
import json
import logging
import pathlib
import platform
# Check for Meta app credentials in environment
META_APP_ID = os.environ.get("META_APP_ID", "")
META_APP_SECRET = os.environ.get("META_APP_SECRET", "")
# Only show warnings about Meta credentials if we're not using Pipeboard
# Check for Pipeboard token in environment
using_pipeboard = bool(os.environ.get("PIPEBOARD_API_TOKEN", ""))
# Print warning if Meta app credentials are not configured and not using Pipeboard
if not using_pipeboard:
if not META_APP_ID:
print("WARNING: META_APP_ID environment variable is not set.")
print("RECOMMENDED: Use Pipeboard authentication by setting PIPEBOARD_API_TOKEN instead.")
print("ALTERNATIVE: For direct Meta authentication, set META_APP_ID to your Meta App ID.")
if not META_APP_SECRET:
print("WARNING: META_APP_SECRET environment variable is not set.")
print("NOTE: This is only needed for direct Meta authentication. Pipeboard authentication doesn't require this.")
print("RECOMMENDED: Use Pipeboard authentication by setting PIPEBOARD_API_TOKEN instead.")
# Configure logging to file
def setup_logging():
"""Set up logging to file for troubleshooting."""
# Get platform-specific path for logs
if platform.system() == "Windows":
base_path = pathlib.Path(os.environ.get("APPDATA", ""))
elif platform.system() == "Darwin": # macOS
base_path = pathlib.Path.home() / "Library" / "Application Support"
else: # Assume Linux/Unix
base_path = pathlib.Path.home() / ".config"
# Create directory if it doesn't exist
log_dir = base_path / "meta-ads-mcp"
log_dir.mkdir(parents=True, exist_ok=True)
log_file = log_dir / "meta_ads_debug.log"
# Configure file logger
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
filename=str(log_file),
filemode='a' # Append mode
)
# Create a logger
logger = logging.getLogger("meta-ads-mcp")
logger.setLevel(logging.DEBUG)
# Log startup information
logger.info(f"Logging initialized. Log file: {log_file}")
logger.info(f"Platform: {platform.system()} {platform.release()}")
logger.info(f"Using Pipeboard authentication: {using_pipeboard}")
return logger
# Create the logger instance to be imported by other modules
logger = setup_logging()
# Global store for ad creative images
ad_creative_images = {}
def extract_creative_image_urls(creative: Dict[str, Any]) -> List[str]:
"""
Extract image URLs from a creative object for direct viewing.
Prioritizes higher quality images over thumbnails.
Args:
creative: Meta Ads creative object
Returns:
List of image URLs found in the creative, prioritized by quality
"""
image_urls = []
# Prioritize higher quality image URLs in this order:
# 1. image_urls_for_viewing (usually highest quality)
# 2. image_url (direct field)
# 3. object_story_spec.link_data.picture (usually full size)
# 4. asset_feed_spec images (multiple high-quality images)
# 5. thumbnail_url (last resort - often profile thumbnail)
# Check for image_urls_for_viewing (highest priority)
if "image_urls_for_viewing" in creative and creative["image_urls_for_viewing"]:
image_urls.extend(creative["image_urls_for_viewing"])
# Check for direct image_url field
if "image_url" in creative and creative["image_url"]:
image_urls.append(creative["image_url"])
# Check object_story_spec for image URLs
if "object_story_spec" in creative:
story_spec = creative["object_story_spec"]
# Check link_data for image fields
if "link_data" in story_spec:
link_data = story_spec["link_data"]
# Check for picture field (usually full size)
if "picture" in link_data and link_data["picture"]:
image_urls.append(link_data["picture"])
# Check for image_url field in link_data
if "image_url" in link_data and link_data["image_url"]:
image_urls.append(link_data["image_url"])
# Check video_data for thumbnail (if present)
if "video_data" in story_spec and "image_url" in story_spec["video_data"]:
image_urls.append(story_spec["video_data"]["image_url"])
# Check asset_feed_spec for multiple images
if "asset_feed_spec" in creative and "images" in creative["asset_feed_spec"]:
for image in creative["asset_feed_spec"]["images"]:
if "url" in image and image["url"]:
image_urls.append(image["url"])
# Check for thumbnail_url field (lowest priority)
if "thumbnail_url" in creative and creative["thumbnail_url"]:
image_urls.append(creative["thumbnail_url"])
# Remove duplicates while preserving order
seen = set()
unique_urls = []
for url in image_urls:
if url not in seen:
seen.add(url)
unique_urls.append(url)
return unique_urls
async def download_image(url: str) -> Optional[bytes]:
"""
Download an image from a URL.
Args:
url: Image URL
Returns:
Image data as bytes if successful, None otherwise
"""
try:
print(f"Attempting to download image from URL: {url}")
# Use minimal headers like curl does
headers = {
"User-Agent": "curl/8.4.0",
"Accept": "*/*"
}
async with httpx.AsyncClient(follow_redirects=True, timeout=30.0) as client:
# Simple GET request just like curl
response = await client.get(url, headers=headers)
# Check response
if response.status_code == 200:
print(f"Successfully downloaded image: {len(response.content)} bytes")
return response.content
else:
print(f"Failed to download image: HTTP {response.status_code}")
return None
except httpx.HTTPStatusError as e:
print(f"HTTP Error when downloading image: {e}")
return None
except httpx.RequestError as e:
print(f"Request Error when downloading image: {e}")
return None
except Exception as e:
print(f"Unexpected error downloading image: {e}")
return None
async def try_multiple_download_methods(url: str) -> Optional[bytes]:
"""
Try multiple methods to download an image, with different approaches for Meta CDN.
Args:
url: Image URL
Returns:
Image data as bytes if successful, None otherwise
"""
# Method 1: Direct download with custom headers
image_data = await download_image(url)
if image_data:
return image_data
print("Direct download failed, trying alternative methods...")
# Method 2: Try adding Facebook cookie simulation
try:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Accept": "image/webp,image/apng,image/*,*/*;q=0.8",
"Cookie": "presence=EDvF3EtimeF1697900316EuserFA21B00112233445566AA0EstateFDutF0CEchF_7bCC" # Fake cookie
}
async with httpx.AsyncClient(follow_redirects=True) as client:
response = await client.get(url, headers=headers, timeout=30.0)
response.raise_for_status()
print(f"Method 2 succeeded with cookie simulation: {len(response.content)} bytes")
return response.content
except Exception as e:
print(f"Method 2 failed: {str(e)}")
# Method 3: Try with session that keeps redirects and cookies
try:
async with httpx.AsyncClient(follow_redirects=True) as client:
# First visit Facebook to get cookies
await client.get("https://www.facebook.com/", timeout=30.0)
# Then try the image URL
response = await client.get(url, timeout=30.0)
response.raise_for_status()
print(f"Method 3 succeeded with Facebook session: {len(response.content)} bytes")
return response.content
except Exception as e:
print(f"Method 3 failed: {str(e)}")
return None
def create_resource_from_image(image_bytes: bytes, resource_id: str, name: str) -> Dict[str, Any]:
"""
Create a resource entry from image bytes.
Args:
image_bytes: Raw image data
resource_id: Unique identifier for the resource
name: Human-readable name for the resource
Returns:
Dictionary with resource information
"""
ad_creative_images[resource_id] = {
"data": image_bytes,
"mime_type": "image/jpeg",
"name": name
}
return {
"resource_id": resource_id,
"resource_uri": f"meta-ads://images/{resource_id}",
"name": name,
"size": len(image_bytes)
}
```
--------------------------------------------------------------------------------
/STREAMABLE_HTTP_SETUP.md:
--------------------------------------------------------------------------------
```markdown
# Streamable HTTP Transport Setup
## Overview
Meta Ads MCP supports **Streamable HTTP Transport**, which allows you to run the server as a standalone HTTP API. This enables direct integration with web applications, custom dashboards, and any system that can make HTTP requests.
## Quick Start
### 1. Start the HTTP Server
```bash
# Basic HTTP server (default: localhost:8080)
python -m meta_ads_mcp --transport streamable-http
# Custom host and port
python -m meta_ads_mcp --transport streamable-http --host 0.0.0.0 --port 9000
```
### 2. Set Authentication
Set your Pipeboard token as an environment variable. This is optional for HTTP transport if you provide the token in the header, but it can be useful for command-line use.
```bash
export PIPEBOARD_API_TOKEN=your_pipeboard_token
```
### 3. Make HTTP Requests
The server accepts JSON-RPC 2.0 requests at the `/mcp` endpoint. Use the `Authorization` header to provide your token.
```bash
curl -X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "Authorization: Bearer your_pipeboard_token" \
-d '{
"jsonrpc": "2.0",
"method": "tools/call",
"id": 1,
"params": {
"name": "get_ad_accounts",
"arguments": {"limit": 5}
}
}'
```
## Configuration Options
### Command Line Arguments
| Argument | Description | Default |
|----------|-------------|---------|
| `--transport` | Transport mode | `stdio` |
| `--host` | Server host address | `localhost` |
| `--port` | Server port | `8080` |
### Examples
```bash
# Local development server
python -m meta_ads_mcp --transport streamable-http --host localhost --port 8080
# Production server (accessible externally)
python -m meta_ads_mcp --transport streamable-http --host 0.0.0.0 --port 8080
# Custom port
python -m meta_ads_mcp --transport streamable-http --port 9000
```
## Authentication
### Primary Method: Bearer Token (Recommended)
1. Sign up at [Pipeboard.co](https://pipeboard.co)
2. Generate an API token at [pipeboard.co/api-tokens](https://pipeboard.co/api-tokens)
3. Include the token in the `Authorization` HTTP header:
```bash
curl -H "Authorization: Bearer your_pipeboard_token" \
-X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-d '{"jsonrpc":"2.0","method":"tools/list","id":1}'
```
### Alternative Method: Direct Meta Token
If you have a Meta Developer App, you can use a direct access token via the `X-META-ACCESS-TOKEN` header. This is less common.
```bash
curl -H "X-META-ACCESS-TOKEN: your_meta_access_token" \
-X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-d '{"jsonrpc":"2.0","method":"tools/list","id":1}'
```
## Available Endpoints
### Server URL Structure
**Base URL**: `http://localhost:8080`
**MCP Endpoint**: `/mcp`
### MCP Protocol Methods
| Method | Description |
|--------|-------------|
| `initialize` | Initialize MCP session and exchange capabilities |
| `tools/list` | Get list of all available Meta Ads tools |
| `tools/call` | Execute a specific tool with parameters |
### Response Format
All responses follow JSON-RPC 2.0 format:
```json
{
"jsonrpc": "2.0",
"id": 1,
"result": {
// Tool response data
}
}
```
## Example Usage
### 1. Initialize Session
```bash
curl -X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "Authorization: Bearer your_token" \
-d '{
"jsonrpc": "2.0",
"method": "initialize",
"id": 1,
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {"roots": {"listChanged": true}},
"clientInfo": {"name": "my-app", "version": "1.0.0"}
}
}'
```
### 2. List Available Tools
```bash
curl -X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "Authorization: Bearer your_token" \
-d '{
"jsonrpc": "2.0",
"method": "tools/list",
"id": 2
}'
```
### 3. Get Ad Accounts
```bash
curl -X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "Authorization: Bearer your_token" \
-d '{
"jsonrpc": "2.0",
"method": "tools/call",
"id": 3,
"params": {
"name": "get_ad_accounts",
"arguments": {"limit": 10}
}
}'
```
### 4. Get Campaign Performance
```bash
curl -X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "Authorization: Bearer your_token" \
-d '{
"jsonrpc": "2.0",
"method": "tools/call",
"id": 4,
"params": {
"name": "get_insights",
"arguments": {
"object_id": "act_701351919139047",
"time_range": "last_30d",
"level": "campaign"
}
}
}'
```
## Client Examples
### Python Client
```python
import requests
import json
class MetaAdsMCPClient:
def __init__(self, base_url="http://localhost:8080", token=None):
self.base_url = base_url
self.endpoint = f"{base_url}/mcp"
self.headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream"
}
if token:
self.headers["Authorization"] = f"Bearer {token}"
def call_tool(self, tool_name, arguments=None):
payload = {
"jsonrpc": "2.0",
"method": "tools/call",
"id": 1,
"params": {"name": tool_name}
}
if arguments:
payload["params"]["arguments"] = arguments
response = requests.post(self.endpoint, headers=self.headers, json=payload)
return response.json()
# Usage
client = MetaAdsMCPClient(token="your_pipeboard_token")
result = client.call_tool("get_ad_accounts", {"limit": 5})
print(json.dumps(result, indent=2))
```
### JavaScript/Node.js Client
```javascript
const axios = require('axios');
class MetaAdsMCPClient {
constructor(baseUrl = 'http://localhost:8080', token = null) {
this.baseUrl = baseUrl;
this.endpoint = `${baseUrl}/mcp`;
this.headers = {
'Content-Type': 'application/json',
'Accept': 'application/json, text/event-stream'
};
if (token) {
this.headers['Authorization'] = `Bearer ${token}`;
}
}
async callTool(toolName, arguments = null) {
const payload = {
jsonrpc: '2.0',
method: 'tools/call',
id: 1,
params: { name: toolName }
};
if (arguments) {
payload.params.arguments = arguments;
}
try {
const response = await axios.post(this.endpoint, payload, { headers: this.headers });
return response.data;
} catch (error) {
return { error: error.message };
}
}
}
// Usage
const client = new MetaAdsMCPClient('http://localhost:8080', 'your_pipeboard_token');
client.callTool('get_ad_accounts', { limit: 5 })
.then(result => console.log(JSON.stringify(result, null, 2)));
```
## Production Deployment
### Security Considerations
1. **Use HTTPS**: In production, run behind a reverse proxy with SSL/TLS
2. **Authentication**: Always use valid Bearer tokens.
3. **Network Security**: Configure firewalls and access controls appropriately
4. **Rate Limiting**: Consider implementing rate limiting for public APIs
### Docker Deployment
```dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY . .
RUN pip install -e .
EXPOSE 8080
CMD ["python", "-m", "meta_ads_mcp", "--transport", "streamable-http", "--host", "0.0.0.0", "--port", "8080"]
```
### Environment Variables
```bash
# For Pipeboard-based authentication. The token will be used for stdio,
# but for HTTP it should be passed in the Authorization header.
export PIPEBOARD_API_TOKEN=your_pipeboard_token
# Optional (for custom Meta apps)
export META_APP_ID=your_app_id
export META_APP_SECRET=your_app_secret
# Optional (for direct Meta token)
export META_ACCESS_TOKEN=your_access_token
```
## Troubleshooting
### Common Issues
1. **Connection Refused**: Ensure the server is running and accessible on the specified port.
2. **Authentication Failed**: Verify your Bearer token is valid and included in the `Authorization` header.
3. **404 Not Found**: Make sure you're using the correct endpoint (`/mcp`).
4. **JSON-RPC Errors**: Check that your request follows the JSON-RPC 2.0 format.
### Debug Mode
Enable verbose logging by setting the log level in your environment if the application supports it, or check the application's logging configuration. The current implementation logs to a file.
### Health Check
Test if the server is running by sending a `tools/list` request:
```bash
curl -X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "Authorization: Bearer your_token" \
-d '{"jsonrpc":"2.0","method":"tools/list","id":1}'
```
## Migration from stdio
If you're currently using stdio transport with MCP clients, you can support both stdio for local clients and HTTP for web applications. The application can only run in one mode at a time, so you may need to run two separate instances if you need both simultaneously.
1. **Keep existing MCP client setup** (Claude Desktop, Cursor, etc.) using stdio.
2. **Add HTTP transport** for web applications and custom integrations by running a separate server instance with the `--transport streamable-http` flag.
3. **Use the same authentication method**:
- For stdio, the `PIPEBOARD_API_TOKEN` environment variable is used.
- For HTTP, pass the token in the `Authorization: Bearer <token>` header.
Both transports access the same Meta Ads functionality and use the same underlying authentication system.
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/authentication.py:
--------------------------------------------------------------------------------
```python
"""Authentication-specific functionality for Meta Ads API.
The Meta Ads MCP server supports three authentication modes:
1. **Development/Local Mode** (default)
- Uses local callback server on localhost:8080+ for OAuth redirect
- Requires META_ADS_DISABLE_CALLBACK_SERVER to NOT be set
- Best for local development and testing
2. **Production with API Token**
- Uses PIPEBOARD_API_TOKEN for server-to-server authentication
- Bypasses OAuth flow entirely
- Best for server deployments with pre-configured tokens
3. **Production OAuth Flow** (NEW)
- Uses Pipeboard OAuth endpoints for dynamic client registration
- Triggered when META_ADS_DISABLE_CALLBACK_SERVER is set but no PIPEBOARD_API_TOKEN
- Supports MCP clients that implement OAuth 2.0 discovery
Environment Variables:
- PIPEBOARD_API_TOKEN: Enables mode 2 (token-based auth)
- META_ADS_DISABLE_CALLBACK_SERVER: Disables local server, enables mode 3
- META_ACCESS_TOKEN: Direct Meta token (fallback)
- META_ADS_DISABLE_LOGIN_LINK: Hard-disables the get_login_link tool; returns a disabled message
"""
import json
from typing import Optional
import asyncio
import os
from .api import meta_api_tool
from . import auth
from .auth import start_callback_server, shutdown_callback_server, auth_manager
from .server import mcp_server
from .utils import logger, META_APP_SECRET
from .pipeboard_auth import pipeboard_auth_manager
# Only register the login link tool if not explicitly disabled
ENABLE_LOGIN_LINK = not bool(os.environ.get("META_ADS_DISABLE_LOGIN_LINK", ""))
async def get_login_link(access_token: Optional[str] = None) -> str:
"""
Get a clickable login link for Meta Ads authentication.
NOTE: This method should only be used if you're using your own Facebook app.
If using Pipeboard authentication (recommended), set the PIPEBOARD_API_TOKEN
environment variable instead (token obtainable via https://pipeboard.co).
Args:
access_token: Meta API access token (optional - will use cached token if not provided)
Returns:
A clickable resource link for Meta authentication
"""
# Check if we're using pipeboard authentication
using_pipeboard = bool(os.environ.get("PIPEBOARD_API_TOKEN", ""))
callback_server_disabled = bool(os.environ.get("META_ADS_DISABLE_CALLBACK_SERVER", ""))
if using_pipeboard:
# Pipeboard token-based authentication
try:
logger.info("Using Pipeboard token-based authentication")
# If an access token was provided, this is likely a test - return success
if access_token:
return json.dumps({
"message": "✅ Authentication Token Provided",
"status": "Using provided access token for authentication",
"token_info": f"Token preview: {access_token[:10]}...",
"authentication_method": "manual_token",
"ready_to_use": "You can now use all Meta Ads MCP tools and commands."
}, indent=2)
# Check if Pipeboard token is working
token = pipeboard_auth_manager.get_access_token()
if token:
return json.dumps({
"message": "✅ Already Authenticated",
"status": "You're successfully authenticated with Meta Ads via Pipeboard!",
"token_info": f"Token preview: {token[:10]}...",
"authentication_method": "pipeboard_token",
"ready_to_use": "You can now use all Meta Ads MCP tools and commands."
}, indent=2)
# Start Pipeboard auth flow
auth_data = pipeboard_auth_manager.initiate_auth_flow()
login_url = auth_data.get('loginUrl')
if login_url:
return json.dumps({
"message": "🔗 Click to Authenticate",
"login_url": login_url,
"markdown_link": f"[🚀 Authenticate with Meta Ads]({login_url})",
"instructions": "Click the link above to complete authentication with Meta Ads.",
"authentication_method": "pipeboard_oauth",
"what_happens_next": "After clicking, you'll be redirected to Meta's authentication page. Once completed, your token will be automatically saved.",
"token_duration": "Your token will be valid for approximately 60 days."
}, indent=2)
else:
return json.dumps({
"message": "❌ Authentication Error",
"error": "Could not generate authentication URL from Pipeboard",
"troubleshooting": [
"Check that your PIPEBOARD_API_TOKEN is valid",
"Ensure the Pipeboard service is accessible",
"Try again in a few moments"
],
"authentication_method": "pipeboard_oauth_failed"
}, indent=2)
except Exception as e:
logger.error(f"Error initiating Pipeboard auth flow: {e}")
return json.dumps({
"message": "❌ Pipeboard Authentication Error",
"error": f"Failed to initiate Pipeboard authentication: {str(e)}",
"troubleshooting": [
"✅ Check that PIPEBOARD_API_TOKEN environment variable is set correctly",
"🌐 Verify that pipeboard.co is accessible from your network",
"🔄 Try refreshing your Pipeboard API token",
"⏰ Wait a moment and try again"
],
"get_help": "Contact support if the issue persists",
"authentication_method": "pipeboard_error"
}, indent=2)
elif callback_server_disabled:
# Production OAuth flow - use Pipeboard OAuth endpoints directly
logger.info("Production OAuth flow - using Pipeboard OAuth endpoints")
return json.dumps({
"message": "🔐 Authentication Required",
"instructions": "Please sign in to your Pipeboard account to authenticate with Meta Ads.",
"sign_in_url": "https://pipeboard.co/auth/signin",
"markdown_link": "[🚀 Sign in to Pipeboard](https://pipeboard.co/auth/signin)",
"what_to_do": "Click the link above to sign in to your Pipeboard account and complete authentication.",
"authentication_method": "production_oauth"
}, indent=2)
else:
# Original Meta authentication flow (development/local)
# Check if we have a cached token
cached_token = auth_manager.get_access_token()
token_status = "No token" if not cached_token else "Valid token"
# If we already have a valid token and none was provided, just return success
if cached_token and not access_token:
logger.info("get_login_link called with existing valid token")
return json.dumps({
"message": "✅ Already Authenticated",
"status": "You're successfully authenticated with Meta Ads!",
"token_info": f"Token preview: {cached_token[:10]}...",
"created_at": auth_manager.token_info.created_at if hasattr(auth_manager, "token_info") else None,
"expires_in": auth_manager.token_info.expires_in if hasattr(auth_manager, "token_info") else None,
"authentication_method": "meta_oauth",
"ready_to_use": "You can now use all Meta Ads MCP tools and commands."
}, indent=2)
# IMPORTANT: Start the callback server first by calling our helper function
# This ensures the server is ready before we provide the URL to the user
logger.info("Starting callback server for authentication")
try:
port = start_callback_server()
logger.info(f"Callback server started on port {port}")
# Generate direct login URL
auth_manager.redirect_uri = f"http://localhost:{port}/callback" # Ensure port is set correctly
logger.info(f"Setting redirect URI to {auth_manager.redirect_uri}")
login_url = auth_manager.get_auth_url()
logger.info(f"Generated login URL: {login_url}")
except Exception as e:
logger.error(f"Failed to start callback server: {e}")
return json.dumps({
"message": "❌ Local Authentication Unavailable",
"error": "Cannot start local callback server for authentication",
"reason": str(e),
"solutions": [
"🌐 Use Pipeboard authentication: Set PIPEBOARD_API_TOKEN environment variable",
"🔑 Use direct token: Set META_ACCESS_TOKEN environment variable",
"🔧 Check if another service is using the required ports"
],
"authentication_method": "meta_oauth_disabled"
}, indent=2)
# Check if we can exchange for long-lived tokens
token_exchange_supported = bool(META_APP_SECRET)
token_duration = "60 days" if token_exchange_supported else "1-2 hours"
# Return a special format that helps the LLM format the response properly
response = {
"message": "🔗 Click to Authenticate",
"login_url": login_url,
"markdown_link": f"[🚀 Authenticate with Meta Ads]({login_url})",
"instructions": "Click the link above to authenticate with Meta Ads.",
"server_info": f"Local callback server running on port {port}",
"token_duration": f"Your token will be valid for approximately {token_duration}",
"authentication_method": "meta_oauth",
"what_happens_next": "After clicking, you'll be redirected to Meta's authentication page. Once completed, your token will be automatically saved.",
"security_note": "This uses a secure local callback server for development purposes."
}
# Wait a moment to ensure the server is fully started
await asyncio.sleep(1)
return json.dumps(response, indent=2)
# Conditionally register as MCP tool only when enabled
if ENABLE_LOGIN_LINK:
get_login_link = mcp_server.tool()(get_login_link)
```
--------------------------------------------------------------------------------
/tests/test_insights_pagination.py:
--------------------------------------------------------------------------------
```python
"""Test pagination functionality for insights endpoint."""
import pytest
import json
from unittest.mock import AsyncMock, patch
from meta_ads_mcp.core.insights import get_insights
class TestInsightsPagination:
"""Test suite for pagination functionality in get_insights"""
@pytest.fixture
def mock_auth_manager(self):
"""Mock for the authentication manager"""
with patch('meta_ads_mcp.core.api.auth_manager') as mock, \
patch('meta_ads_mcp.core.auth.get_current_access_token') as mock_get_token:
# Mock a valid access token
mock.get_current_access_token.return_value = "test_access_token"
mock.is_token_valid.return_value = True
mock.app_id = "test_app_id"
mock_get_token.return_value = "test_access_token"
yield mock
@pytest.fixture
def valid_account_id(self):
return "act_701351919139047"
@pytest.fixture
def mock_paginated_response_page1(self):
"""Mock first page of paginated response"""
return {
"data": [
{
"campaign_id": "campaign_1",
"campaign_name": "Test Campaign 1",
"spend": "100.50",
"impressions": "1000",
"clicks": "50"
},
{
"campaign_id": "campaign_2",
"campaign_name": "Test Campaign 2",
"spend": "200.75",
"impressions": "2000",
"clicks": "100"
}
],
"paging": {
"cursors": {
"before": "before_cursor_1",
"after": "after_cursor_1"
},
"next": "https://graph.facebook.com/v20.0/act_123/insights?after=after_cursor_1&limit=2"
}
}
@pytest.fixture
def mock_paginated_response_page2(self):
"""Mock second page of paginated response"""
return {
"data": [
{
"campaign_id": "campaign_3",
"campaign_name": "Test Campaign 3",
"spend": "150.25",
"impressions": "1500",
"clicks": "75"
}
],
"paging": {
"cursors": {
"before": "before_cursor_2",
"after": "after_cursor_2"
}
}
}
@pytest.mark.asyncio
async def test_insights_with_limit_parameter(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
"""Test that limit parameter is properly passed to API"""
with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
mock_api_request.return_value = mock_paginated_response_page1
result = await get_insights(
object_id=valid_account_id,
level="campaign",
time_range="last_30d",
limit=2
)
# Verify the API was called with correct parameters
mock_api_request.assert_called_once()
call_args = mock_api_request.call_args
# Check that limit is included in params
params = call_args[0][2]
assert params["limit"] == 2
assert params["level"] == "campaign"
assert params["date_preset"] == "last_30d"
# Verify the response structure
result_data = json.loads(result)
assert "data" in result_data
assert len(result_data["data"]) == 2
assert "paging" in result_data
@pytest.mark.asyncio
async def test_insights_with_after_cursor(self, mock_auth_manager, valid_account_id, mock_paginated_response_page2):
"""Test that after cursor is properly passed to API for pagination"""
with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
mock_api_request.return_value = mock_paginated_response_page2
after_cursor = "after_cursor_1"
result = await get_insights(
object_id=valid_account_id,
level="campaign",
time_range="last_30d",
limit=10,
after=after_cursor
)
# Verify the API was called with correct parameters
mock_api_request.assert_called_once()
call_args = mock_api_request.call_args
# Check that after cursor is included in params
params = call_args[0][2]
assert params["after"] == after_cursor
assert params["limit"] == 10
# Verify the response structure
result_data = json.loads(result)
assert "data" in result_data
assert len(result_data["data"]) == 1
@pytest.mark.asyncio
async def test_insights_default_limit(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
"""Test that default limit is 25 when not specified"""
with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
mock_api_request.return_value = mock_paginated_response_page1
result = await get_insights(
object_id=valid_account_id,
level="campaign"
)
# Verify the API was called with default limit
mock_api_request.assert_called_once()
call_args = mock_api_request.call_args
params = call_args[0][2]
assert params["limit"] == 25 # Default value
@pytest.mark.asyncio
async def test_insights_without_after_cursor(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
"""Test that after parameter is not included when empty"""
with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
mock_api_request.return_value = mock_paginated_response_page1
result = await get_insights(
object_id=valid_account_id,
level="campaign",
after="" # Empty after cursor
)
# Verify the API was called without after parameter
mock_api_request.assert_called_once()
call_args = mock_api_request.call_args
params = call_args[0][2]
assert "after" not in params # Should not be included when empty
@pytest.mark.asyncio
async def test_insights_pagination_with_custom_time_range(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
"""Test pagination works with custom time range"""
with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
mock_api_request.return_value = mock_paginated_response_page1
custom_time_range = {"since": "2024-01-01", "until": "2024-01-31"}
result = await get_insights(
object_id=valid_account_id,
level="campaign",
time_range=custom_time_range,
limit=5,
after="test_cursor"
)
# Verify the API was called with correct parameters
mock_api_request.assert_called_once()
call_args = mock_api_request.call_args
params = call_args[0][2]
assert params["limit"] == 5
assert params["after"] == "test_cursor"
assert params["time_range"] == json.dumps(custom_time_range)
@pytest.mark.asyncio
async def test_insights_pagination_with_breakdown(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
"""Test pagination works with breakdown parameter"""
with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
mock_api_request.return_value = mock_paginated_response_page1
result = await get_insights(
object_id=valid_account_id,
level="campaign",
breakdown="age",
limit=10,
after="test_cursor_2"
)
# Verify the API was called with correct parameters
mock_api_request.assert_called_once()
call_args = mock_api_request.call_args
params = call_args[0][2]
assert params["limit"] == 10
assert params["after"] == "test_cursor_2"
assert params["breakdowns"] == "age"
@pytest.mark.asyncio
async def test_insights_large_limit_value(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
"""Test that large limit values are accepted (API will enforce its own limits)"""
with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
mock_api_request.return_value = mock_paginated_response_page1
result = await get_insights(
object_id=valid_account_id,
level="campaign",
limit=1000 # Large limit - API will enforce its own max
)
# Verify the API was called with the large limit
mock_api_request.assert_called_once()
call_args = mock_api_request.call_args
params = call_args[0][2]
assert params["limit"] == 1000
@pytest.mark.asyncio
async def test_insights_paging_response_structure(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
"""Test that paging information is preserved in the response"""
with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
mock_api_request.return_value = mock_paginated_response_page1
result = await get_insights(
object_id=valid_account_id,
level="campaign",
limit=2
)
# Verify the response includes paging information
result_data = json.loads(result)
assert "data" in result_data
assert "paging" in result_data
assert "cursors" in result_data["paging"]
assert "after" in result_data["paging"]["cursors"]
assert "next" in result_data["paging"]
```
--------------------------------------------------------------------------------
/tests/test_duplication.py:
--------------------------------------------------------------------------------
```python
"""Tests for the duplication module."""
import os
import json
import pytest
from unittest.mock import patch, AsyncMock, Mock
from meta_ads_mcp.core.duplication import ENABLE_DUPLICATION
def test_duplication_disabled_by_default():
"""Test that duplication is disabled by default."""
# Test with no environment variable set
with patch.dict(os.environ, {}, clear=True):
from meta_ads_mcp.core import duplication
# When imported fresh, it should be disabled
assert not duplication.ENABLE_DUPLICATION
def test_duplication_enabled_with_env_var():
"""Test that duplication is enabled when environment variable is set."""
with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
# Need to reload the module to pick up the new environment variable
import importlib
from meta_ads_mcp.core import duplication
importlib.reload(duplication)
assert duplication.ENABLE_DUPLICATION
@pytest.mark.asyncio
async def test_forward_duplication_request_no_pipeboard_token():
"""Test that _forward_duplication_request handles missing Pipeboard token."""
from meta_ads_mcp.core.duplication import _forward_duplication_request
# Mock the auth integration to return no Pipeboard token but a Facebook token
with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
mock_auth.get_pipeboard_token.return_value = None # No Pipeboard token
mock_auth.get_auth_token.return_value = "facebook_token" # Has Facebook token
result = await _forward_duplication_request("campaign", "123456789", None, {})
result_json = json.loads(result)
assert result_json["error"] == "authentication_required"
assert "Pipeboard API token not found" in result_json["message"]
@pytest.mark.asyncio
async def test_forward_duplication_request_no_facebook_token():
"""Test that _forward_duplication_request handles missing Facebook token."""
from meta_ads_mcp.core.duplication import _forward_duplication_request
# Mock the auth integration to return Pipeboard token but no Facebook token
with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
mock_auth.get_pipeboard_token.return_value = "pipeboard_token" # Has Pipeboard token
mock_auth.get_auth_token.return_value = None # No Facebook token
# Mock get_current_access_token to also return None
with patch("meta_ads_mcp.core.auth.get_current_access_token") as mock_get_token:
mock_get_token.return_value = None
result = await _forward_duplication_request("campaign", "123456789", None, {})
result_json = json.loads(result)
assert result_json["error"] == "authentication_required"
assert "Meta Ads access token not found" in result_json["message"]
@pytest.mark.asyncio
async def test_forward_duplication_request_with_both_tokens():
"""Test that _forward_duplication_request makes HTTP request with dual headers."""
from meta_ads_mcp.core.duplication import _forward_duplication_request
mock_response = Mock()
mock_response.status_code = 403
mock_response.json.return_value = {"error": "premium_feature"}
# Mock the auth integration to return both tokens
with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
mock_auth.get_auth_token.return_value = "facebook_token"
with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
result = await _forward_duplication_request("campaign", "123456789", None, {
"name_suffix": " - Test"
})
result_json = json.loads(result)
# Should return premium feature message for 403 response
assert result_json["error"] == "premium_feature_required"
assert "premium feature" in result_json["message"]
# Verify the HTTP request was made with correct parameters
mock_client.return_value.__aenter__.return_value.post.assert_called_once()
call_args = mock_client.return_value.__aenter__.return_value.post.call_args
# Check URL
assert call_args[0][0] == "https://mcp.pipeboard.co/api/meta/duplicate/campaign/123456789"
# Check dual headers (the key change!)
headers = call_args[1]["headers"]
assert headers["Authorization"] == "Bearer facebook_token" # Facebook token for Meta API
assert headers["X-Pipeboard-Token"] == "pipeboard_token" # Pipeboard token for auth
assert headers["Content-Type"] == "application/json"
# Check JSON payload
json_payload = call_args[1]["json"]
assert json_payload == {"name_suffix": " - Test"}
@pytest.mark.asyncio
async def test_forward_duplication_request_with_provided_access_token():
"""Test that provided access_token parameter is used when available."""
from meta_ads_mcp.core.duplication import _forward_duplication_request
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"success": True, "new_campaign_id": "987654321"}
# Mock the auth integration to return Pipeboard token but no Facebook token in context
with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
mock_auth.get_auth_token.return_value = None # No Facebook token in context
with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
# Provide access_token as parameter
result = await _forward_duplication_request("campaign", "123456789", "provided_facebook_token", {})
result_json = json.loads(result)
# Should succeed
assert result_json["success"] is True
# Verify the HTTP request used the provided token
call_args = mock_client.return_value.__aenter__.return_value.post.call_args
headers = call_args[1]["headers"]
assert headers["Authorization"] == "Bearer provided_facebook_token"
assert headers["X-Pipeboard-Token"] == "pipeboard_token"
@pytest.mark.asyncio
async def test_duplicate_campaign_function_available_when_enabled():
"""Test that duplicate_campaign function is available when feature is enabled."""
with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
# Reload module to pick up environment variable
import importlib
from meta_ads_mcp.core import duplication
importlib.reload(duplication)
# Function should be available
assert hasattr(duplication, 'duplicate_campaign')
# Test that it calls the forwarding function
with patch("meta_ads_mcp.core.duplication._forward_duplication_request") as mock_forward:
mock_forward.return_value = '{"success": true}'
result = await duplication.duplicate_campaign("123456789", access_token="test_token")
mock_forward.assert_called_once_with(
"campaign",
"123456789",
"test_token",
{
"name_suffix": " - Copy",
"include_ad_sets": True,
"include_ads": True,
"include_creatives": True,
"copy_schedule": False,
"new_daily_budget": None,
"new_status": "PAUSED"
}
)
def test_get_estimated_components():
"""Test the _get_estimated_components helper function."""
from meta_ads_mcp.core.duplication import _get_estimated_components
# Test campaign with all components
campaign_result = _get_estimated_components("campaign", {
"include_ad_sets": True,
"include_ads": True,
"include_creatives": True
})
assert campaign_result["campaigns"] == 1
assert "ad_sets" in campaign_result
assert "ads" in campaign_result
assert "creatives" in campaign_result
# Test adset
adset_result = _get_estimated_components("adset", {"include_ads": True})
assert adset_result["ad_sets"] == 1
assert "ads" in adset_result
# Test creative only
creative_result = _get_estimated_components("creative", {})
assert creative_result == {"creatives": 1}
@pytest.mark.asyncio
async def test_dual_header_authentication_integration():
"""Test that the dual-header authentication works end-to-end."""
from meta_ads_mcp.core.duplication import _forward_duplication_request
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"success": True,
"new_campaign_id": "987654321",
"subscription": {"status": "active"}
}
# Test the complete dual-header flow
with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
mock_auth.get_pipeboard_token.return_value = "pb_token_12345"
mock_auth.get_auth_token.return_value = "fb_token_67890"
with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
result = await _forward_duplication_request("adset", "456789", None, {
"target_campaign_id": "123456",
"include_ads": True
})
result_json = json.loads(result)
# Should succeed
assert result_json["success"] is True
assert result_json["new_campaign_id"] == "987654321"
# Verify correct endpoint was called
call_args = mock_client.return_value.__aenter__.return_value.post.call_args
assert "adset/456789" in call_args[0][0]
# Verify dual headers were sent correctly
headers = call_args[1]["headers"]
assert headers["Authorization"] == "Bearer fb_token_67890"
assert headers["X-Pipeboard-Token"] == "pb_token_12345"
# Verify payload
payload = call_args[1]["json"]
assert payload["target_campaign_id"] == "123456"
assert payload["include_ads"] is True
```
--------------------------------------------------------------------------------
/LOCAL_INSTALLATION.md:
--------------------------------------------------------------------------------
```markdown
# Meta Ads MCP - Local Installation Guide
This guide covers everything you need to know about installing and running Meta Ads MCP locally on your machine. For the easier Remote MCP option, **[🚀 get started here](https://pipeboard.co)**.
## Table of Contents
- [Prerequisites](#prerequisites)
- [Installation Methods](#installation-methods)
- [Authentication Setup](#authentication-setup)
- [MCP Client Configuration](#mcp-client-configuration)
- [Development Installation](#development-installation)
- [Privacy and Security](#privacy-and-security)
- [Testing and Verification](#testing-and-verification)
- [Debugging and Logs](#debugging-and-logs)
- [Troubleshooting](#troubleshooting)
- [Advanced Configuration](#advanced-configuration)
## Prerequisites
- **Python 3.8 or higher**
- **[uv](https://docs.astral.sh/uv/) package manager** (recommended) or pip
- **Meta Ads account** with appropriate permissions
- **MCP-compatible client** (Claude Desktop, Cursor, Cherry Studio, etc.)
## Installation Methods
### Method 1: Using uvx (Recommended)
```bash
# Install via uvx (automatically handles dependencies)
uvx meta-ads-mcp
```
### Method 2: Using pip
```bash
# Install via pip
pip install meta-ads-mcp
```
### Method 3: Development Installation
```bash
# Clone the repository
git clone https://github.com/pipeboard-co/meta-ads-mcp.git
cd meta-ads-mcp
# Install in development mode
uv pip install -e .
# Or with pip
pip install -e .
```
## Authentication Setup
You have two authentication options:
### Option 1: Pipeboard Authentication (Recommended)
This is the easiest method that handles all OAuth complexity for you:
1. **Sign up to Pipeboard**
- Visit [Pipeboard.co](https://pipeboard.co)
- Create an account
2. **Generate API Token**
- Go to [pipeboard.co/api-tokens](https://pipeboard.co/api-tokens)
- Generate a new API token
- Copy the token securely
3. **Set Environment Variable**
```bash
# On macOS/Linux
export PIPEBOARD_API_TOKEN=your_pipeboard_token_here
# On Windows (Command Prompt)
set PIPEBOARD_API_TOKEN=your_pipeboard_token_here
# On Windows (PowerShell)
$env:PIPEBOARD_API_TOKEN="your_pipeboard_token_here"
```
4. **Make it Persistent**
Add to your shell profile (`.bashrc`, `.zshrc`, etc.):
```bash
echo 'export PIPEBOARD_API_TOKEN=your_pipeboard_token_here' >> ~/.bashrc
source ~/.bashrc
```
### Option 2: Custom Meta App
If you prefer to use your own Meta Developer App, see [CUSTOM_META_APP.md](CUSTOM_META_APP.md) for detailed instructions.
## MCP Client Configuration
### Claude Desktop
Add to your `claude_desktop_config.json`:
**Location:**
- **macOS**: `~/Library/Application Support/Claude/claude_desktop_config.json`
- **Windows**: `%APPDATA%\Claude\claude_desktop_config.json`
**Configuration:**
```json
{
"mcpServers": {
"meta-ads": {
"command": "uvx",
"args": ["meta-ads-mcp"],
"env": {
"PIPEBOARD_API_TOKEN": "your_pipeboard_token"
}
}
}
}
```
### Cursor
Add to your `~/.cursor/mcp.json`:
```json
{
"mcpServers": {
"meta-ads": {
"command": "uvx",
"args": ["meta-ads-mcp"],
"env": {
"PIPEBOARD_API_TOKEN": "your_pipeboard_token"
}
}
}
}
```
### Cherry Studio
In Cherry Studio settings, add a new MCP server:
- **Name**: Meta Ads MCP
- **Command**: `uvx`
- **Arguments**: `["meta-ads-mcp"]`
- **Environment Variables**: `PIPEBOARD_API_TOKEN=your_pipeboard_token`
## Development Installation
### Setting Up Development Environment
```bash
# Clone the repository
git clone https://github.com/pipeboard-co/meta-ads-mcp.git
cd meta-ads-mcp
# Create virtual environment (optional but recommended)
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install in development mode with dependencies
uv pip install -e .
# Install development dependencies
uv pip install -e ".[dev]" # If dev dependencies are defined
```
### Running from Source
```bash
# Set environment variable
export PIPEBOARD_API_TOKEN=your_token
# Run directly
python -m meta_ads_mcp
# Or if installed in development mode
meta-ads-mcp
```
### Testing Your Installation
```bash
# Test the installation
python -c "import meta_ads_mcp; print('Installation successful!')"
# Test MCP server startup
meta-ads-mcp --help
```
## Privacy and Security
### Token Storage and Caching
Meta Ads MCP follows security best practices:
1. **Secure Token Cache Location**:
- **Windows**: `%APPDATA%\meta-ads-mcp\token_cache.json`
- **macOS**: `~/Library/Application Support/meta-ads-mcp/token_cache.json`
- **Linux**: `~/.config/meta-ads-mcp/token_cache.json`
2. **Automatic Token Management**:
- Tokens are cached securely after first authentication
- You don't need to provide access tokens for each command
- Tokens are automatically refreshed when needed
3. **Environment Variable Security**:
- `PIPEBOARD_API_TOKEN` should be kept secure
- Don't commit tokens to version control
- Use environment files (`.env`) for local development
### Security Best Practices
```bash
# Create a .env file for local development (never commit this)
echo "PIPEBOARD_API_TOKEN=your_token_here" > .env
# Add .env to .gitignore
echo ".env" >> .gitignore
# Load environment variables from .env
source .env
```
## Testing and Verification
### Basic Functionality Test
Once installed and configured, test with your MCP client:
1. **Verify Account Access**
```
Ask your LLM: "Use mcp_meta_ads_get_ad_accounts to show my Meta ad accounts"
```
2. **Check Account Details**
```
Ask your LLM: "Get details for account act_XXXXXXXXX using mcp_meta_ads_get_account_info"
```
3. **List Campaigns**
```
Ask your LLM: "Show me my active campaigns using mcp_meta_ads_get_campaigns"
```
### Manual Testing with Python
```python
# Test authentication
from meta_ads_mcp.core.auth import get_access_token
try:
token = get_access_token()
print("Authentication successful!")
print(f"Token starts with: {token[:10]}...")
except Exception as e:
print(f"Authentication failed: {e}")
```
### Testing with MCP Client
When using Meta Ads MCP with an LLM interface:
1. Ensure the `PIPEBOARD_API_TOKEN` environment variable is set
2. Verify account access by calling `mcp_meta_ads_get_ad_accounts`
3. Check specific account details with `mcp_meta_ads_get_account_info`
4. Test campaign retrieval with `mcp_meta_ads_get_campaigns`
## Debugging and Logs
### Log File Locations
Debug logs are automatically created in platform-specific locations:
- **macOS**: `~/Library/Application\ Support/meta-ads-mcp/meta_ads_debug.log`
- **Windows**: `%APPDATA%\meta-ads-mcp\meta_ads_debug.log`
- **Linux**: `~/.config/meta-ads-mcp/meta_ads_debug.log`
### Enabling Debug Mode
```bash
# Set debug environment variable
export META_ADS_DEBUG=true
# Run with verbose output
meta-ads-mcp --verbose
```
### Viewing Logs
```bash
# On macOS/Linux
tail -f ~/Library/Application\ Support/meta-ads-mcp/meta_ads_debug.log
# On Windows
type %APPDATA%\meta-ads-mcp\meta_ads_debug.log
```
### Common Debug Commands
```bash
# Check if MCP server starts correctly
meta-ads-mcp --test-connection
# Verify environment variables
echo $PIPEBOARD_API_TOKEN
# Test Pipeboard authentication
python -c "
from meta_ads_mcp.core.pipeboard_auth import test_auth
test_auth()
"
```
## Troubleshooting
### Authentication Issues
#### Problem: "PIPEBOARD_API_TOKEN not set"
```bash
# Solution: Set the environment variable
export PIPEBOARD_API_TOKEN=your_token_here
# Verify it's set
echo $PIPEBOARD_API_TOKEN
```
#### Problem: "Invalid Pipeboard token"
1. Check your token at [pipeboard.co/api-tokens](https://pipeboard.co/api-tokens)
2. Regenerate if necessary
3. Update your environment variable
#### Problem: "Authentication failed"
```bash
# Clear cached tokens and retry
rm -rf ~/.config/meta-ads-mcp/token_cache.json # Linux
rm -rf ~/Library/Application\ Support/meta-ads-mcp/token_cache.json # macOS
# Force re-authentication
python test_pipeboard_auth.py --force-login
```
### Installation Issues
#### Problem: "Command not found: uvx"
```bash
# Install uv first
curl -LsSf https://astral.sh/uv/install.sh | sh
# Or use pip
pip install meta-ads-mcp
```
#### Problem: "Permission denied"
```bash
# Use user installation
pip install --user meta-ads-mcp
# Or use virtual environment
python -m venv venv
source venv/bin/activate
pip install meta-ads-mcp
```
#### Problem: "Python version incompatible"
```bash
# Check Python version
python --version
# Update to Python 3.8+
# Use pyenv or your system's package manager
```
### Runtime Issues
#### Problem: "Failed to connect to Meta API"
1. Check internet connection
2. Verify Meta API status
3. Check rate limits
4. Ensure account permissions
#### Problem: "MCP client can't find server"
1. Verify the command path in your MCP client config
2. Check environment variables are set in the client
3. Test the command manually in terminal
#### Problem: "SSL/TLS errors"
```bash
# Update certificates
pip install --upgrade certifi
# Or use system certificates
export SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt
```
### API Errors
#### Problem: "Insufficient permissions"
- Ensure your Meta account has access to the ad accounts
- Check if your Pipeboard token has the right scopes
- Verify account roles in Meta Business Manager
#### Problem: "Rate limit exceeded"
- Wait before retrying
- Reduce request frequency
- Check if multiple instances are running
#### Problem: "Account not found"
- Verify account ID format (should be `act_XXXXXXXXX`)
- Check account access permissions
- Ensure account is active
### Performance Issues
#### Problem: "Slow response times"
```bash
# Check network latency
ping graph.facebook.com
# Clear cache
rm -rf ~/.config/meta-ads-mcp/token_cache.json
# Check system resources
top # or htop on Linux/macOS
```
## Advanced Configuration
### Custom Configuration File
Create `~/.config/meta-ads-mcp/config.json`:
```json
{
"api_version": "v21.0",
"timeout": 30,
"max_retries": 3,
"debug": false,
"cache_duration": 3600
}
```
### Environment Variables
```bash
# API Configuration
export META_API_VERSION=v21.0
export META_API_TIMEOUT=30
export META_ADS_DEBUG=true
# Cache Configuration
export META_ADS_CACHE_DIR=/custom/cache/path
export META_ADS_CACHE_DURATION=3600
# Pipeboard Configuration
export PIPEBOARD_API_BASE=https://api.pipeboard.co
export PIPEBOARD_API_TOKEN=your_token_here
```
### Transport Configuration
Meta Ads MCP uses **stdio transport** by default. For HTTP transport:
See [STREAMABLE_HTTP_SETUP.md](STREAMABLE_HTTP_SETUP.md) for streamable HTTP transport configuration.
### Custom Meta App Integration
For advanced users who want to use their own Meta Developer App:
1. Follow [CUSTOM_META_APP.md](CUSTOM_META_APP.md) guide
2. Set up OAuth flow
3. Configure environment variables:
```bash
export META_APP_ID=your_app_id
export META_APP_SECRET=your_app_secret
export META_REDIRECT_URI=your_redirect_uri
```
## Getting Help
If you're still experiencing issues:
1. **Check the logs** for detailed error messages
2. **Search existing issues** on GitHub
3. **Join our Discord** at [discord.gg/YzMwQ8zrjr](https://discord.gg/YzMwQ8zrjr)
4. **Email support** at [email protected]
5. **Consider Remote MCP** at [pipeboard.co](https://pipeboard.co) as an alternative
---
**Quick Alternative**: If local installation is causing issues, try our [Remote MCP service](https://pipeboard.co) - no local setup required!
```
--------------------------------------------------------------------------------
/tests/test_account_search.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""
Focused Account Search Test for Meta Ads MCP
This test validates that the search tool correctly finds and returns
account data for known test accounts.
Expected test accounts:
- act_4891437610982483 (Yves Junqueira)
- act_701351919139047 (Injury Payouts)
"""
import requests
import json
import os
import sys
from typing import Dict, Any, List
# Load environment variables from .env file
try:
from dotenv import load_dotenv
load_dotenv()
print("✅ Loaded environment variables from .env file")
except ImportError:
print("⚠️ python-dotenv not installed, using system environment variables only")
class AccountSearchTester:
"""Test suite focused on account search functionality"""
def __init__(self, base_url: str = "http://localhost:8080"):
self.base_url = base_url.rstrip('/')
self.endpoint = f"{self.base_url}/mcp/"
self.request_id = 1
# Expected test data
self.expected_accounts = [
{
"id": "act_4891437610982483",
"name": "Yves Junqueira",
"account_id": "4891437610982483"
},
{
"id": "act_701351919139047",
"name": "Injury Payouts",
"account_id": "701351919139047"
}
]
def _make_request(self, method: str, params: Dict[str, Any] = None,
headers: Dict[str, str] = None) -> Dict[str, Any]:
"""Make a JSON-RPC request to the MCP server"""
default_headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
"User-Agent": "Account-Search-Test-Client/1.0"
}
if headers:
default_headers.update(headers)
payload = {
"jsonrpc": "2.0",
"method": method,
"id": self.request_id
}
if params:
payload["params"] = params
try:
response = requests.post(
self.endpoint,
headers=default_headers,
json=payload,
timeout=10
)
self.request_id += 1
return {
"status_code": response.status_code,
"headers": dict(response.headers),
"json": response.json() if response.status_code == 200 else None,
"text": response.text,
"success": response.status_code == 200
}
except requests.exceptions.RequestException as e:
return {
"status_code": 0,
"headers": {},
"json": None,
"text": str(e),
"success": False,
"error": str(e)
}
def test_search_accounts(self) -> Dict[str, Any]:
"""Test searching for accounts with various queries"""
queries_to_test = [
"accounts",
"ad accounts",
"meta accounts",
"Yves",
"Injury Payouts"
]
results = {}
for query in queries_to_test:
print(f"\n🔍 Testing search query: '{query}'")
result = self._make_request("tools/call", {
"name": "search",
"arguments": {"query": query}
})
if not result["success"]:
results[query] = {
"success": False,
"error": result.get("text", "Unknown error")
}
print(f"❌ Search failed: {result.get('text', 'Unknown error')}")
continue
# Parse the tool response
response_data = result["json"]["result"]
content = response_data.get("content", [{}])[0].get("text", "")
try:
parsed_content = json.loads(content)
ids = parsed_content.get("ids", [])
# Check for expected account IDs
expected_account_ids = [f"account:{acc['id']}" for acc in self.expected_accounts]
found_account_ids = [id for id in ids if id.startswith("account:")]
results[query] = {
"success": True,
"ids": ids,
"account_ids": found_account_ids,
"found_expected_accounts": len([id for id in expected_account_ids if id in found_account_ids]),
"total_expected": len(expected_account_ids),
"raw_content": parsed_content
}
print(f"✅ Found {len(found_account_ids)} account IDs: {found_account_ids}")
print(f"📊 Expected accounts found: {results[query]['found_expected_accounts']}/{results[query]['total_expected']}")
print(f"🔍 Raw response: {json.dumps(parsed_content, indent=2)}")
except json.JSONDecodeError:
results[query] = {
"success": False,
"error": "Search tool did not return valid JSON",
"raw_content": content
}
print(f"❌ Invalid JSON response: {content}")
return results
def test_fetch_account(self, account_id: str) -> Dict[str, Any]:
"""Test fetching a specific account by ID"""
print(f"\n🔍 Testing fetch for account: {account_id}")
result = self._make_request("tools/call", {
"name": "fetch",
"arguments": {"id": account_id}
})
if not result["success"]:
return {
"success": False,
"error": result.get("text", "Unknown error")
}
# Parse the tool response
response_data = result["json"]["result"]
content = response_data.get("content", [{}])[0].get("text", "")
try:
parsed_content = json.loads(content)
# Validate required fields for OpenAI MCP compliance
required_fields = ["id", "title", "text"]
has_required_fields = all(field in parsed_content for field in required_fields)
result_data = {
"success": True,
"record": parsed_content,
"has_required_fields": has_required_fields,
"missing_fields": [field for field in required_fields if field not in parsed_content]
}
if has_required_fields:
print(f"✅ Successfully fetched account with all required fields")
print(f" Title: {parsed_content.get('title', 'N/A')}")
print(f" ID: {parsed_content.get('id', 'N/A')}")
else:
print(f"⚠️ Account fetched but missing required fields: {result_data['missing_fields']}")
return result_data
except json.JSONDecodeError:
return {
"success": False,
"error": "Fetch tool did not return valid JSON",
"raw_content": content
}
def run_account_search_tests(self) -> bool:
"""Run comprehensive account search tests"""
print("🚀 Meta Ads Account Search Test Suite")
print("="*50)
# Check server availability
try:
response = requests.get(f"{self.base_url}/", timeout=5)
server_running = response.status_code in [200, 404]
except:
server_running = False
if not server_running:
print("❌ Server is not running at", self.base_url)
print(" Please start the server with:")
print(" python3 -m meta_ads_mcp --transport streamable-http --port 8080")
return False
print("✅ Server is running")
print("🔐 Using implicit authentication from server")
# Test 0: First try get_ad_accounts to see if we can get raw data
print("\n" + "="*50)
print("📋 PHASE 0: Testing Direct Account Access")
print("="*50)
account_result = self._make_request("tools/call", {
"name": "get_ad_accounts",
"arguments": {
"user_id": "me",
"parameters": json.dumps({"limit": 5})
}
})
if account_result["success"]:
response_data = account_result["json"]["result"]
content = response_data.get("content", [{}])[0].get("text", "")
try:
account_data = json.loads(content)
print(f"✅ get_ad_accounts returned: {json.dumps(account_data, indent=2)}")
except:
print(f"⚠️ get_ad_accounts raw response: {content}")
else:
print(f"❌ get_ad_accounts failed: {account_result.get('text', 'Unknown error')}")
# Test 1: Search for accounts
print("\n" + "="*50)
print("📋 PHASE 1: Testing Account Search")
print("="*50)
search_results = self.test_search_accounts()
# Find the best search result that returned accounts
best_search = None
for query, result in search_results.items():
if result.get("success") and result.get("account_ids"):
best_search = result
break
if not best_search:
print("\n❌ No search queries returned account IDs")
print("📊 Search Results Summary:")
for query, result in search_results.items():
if result.get("success"):
print(f" '{query}': {len(result.get('ids', []))} total IDs, {len(result.get('account_ids', []))} account IDs")
else:
print(f" '{query}': FAILED - {result.get('error', 'Unknown error')}")
return False
print(f"\n✅ Found accounts in search results")
account_ids = best_search["account_ids"]
print(f"📋 Account IDs found: {account_ids}")
# Test 2: Fetch account details
print("\n" + "="*50)
print("📋 PHASE 2: Testing Account Fetch")
print("="*50)
fetch_success = True
for account_id in account_ids[:2]: # Test first 2 accounts
fetch_result = self.test_fetch_account(account_id)
if not fetch_result["success"]:
print(f"❌ Failed to fetch {account_id}: {fetch_result.get('error', 'Unknown error')}")
fetch_success = False
elif not fetch_result["has_required_fields"]:
print(f"⚠️ {account_id} missing required fields: {fetch_result['missing_fields']}")
fetch_success = False
# Final assessment
print("\n" + "="*50)
print("📊 FINAL RESULTS")
print("="*50)
if fetch_success and account_ids:
print("✅ Account search and fetch workflow: SUCCESS")
print(f" • Found {len(account_ids)} accounts")
print(f" • All fetched accounts have required fields")
print(f" • OpenAI MCP compliance: PASSED")
return True
else:
print("❌ Account search and fetch workflow: FAILED")
if not account_ids:
print(" • Issue: No account IDs returned by search")
if not fetch_success:
print(" • Issue: Some accounts failed to fetch or missing required fields")
return False
def main():
"""Main test execution"""
tester = AccountSearchTester()
success = tester.run_account_search_tests()
if success:
print("\n🎉 All account search tests passed!")
else:
print("\n⚠️ Some account search tests failed - see details above")
sys.exit(0 if success else 1)
if __name__ == "__main__":
main()
```