This is page 2 of 6. Use http://codebase.md/nictuku/meta-ads-mcp?lines=true&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
--------------------------------------------------------------------------------
/meta_ads_mcp/core/authentication.py:
--------------------------------------------------------------------------------
```python
1 | """Authentication-specific functionality for Meta Ads API.
2 |
3 | The Meta Ads MCP server supports three authentication modes:
4 |
5 | 1. **Development/Local Mode** (default)
6 | - Uses local callback server on localhost:8080+ for OAuth redirect
7 | - Requires META_ADS_DISABLE_CALLBACK_SERVER to NOT be set
8 | - Best for local development and testing
9 |
10 | 2. **Production with API Token**
11 | - Uses PIPEBOARD_API_TOKEN for server-to-server authentication
12 | - Bypasses OAuth flow entirely
13 | - Best for server deployments with pre-configured tokens
14 |
15 | 3. **Production OAuth Flow** (NEW)
16 | - Uses Pipeboard OAuth endpoints for dynamic client registration
17 | - Triggered when META_ADS_DISABLE_CALLBACK_SERVER is set but no PIPEBOARD_API_TOKEN
18 | - Supports MCP clients that implement OAuth 2.0 discovery
19 |
20 | Environment Variables:
21 | - PIPEBOARD_API_TOKEN: Enables mode 2 (token-based auth)
22 | - META_ADS_DISABLE_CALLBACK_SERVER: Disables local server, enables mode 3
23 | - META_ACCESS_TOKEN: Direct Meta token (fallback)
24 | - META_ADS_DISABLE_LOGIN_LINK: Hard-disables the get_login_link tool; returns a disabled message
25 | """
26 |
27 | import json
28 | from typing import Optional
29 | import asyncio
30 | import os
31 | from .api import meta_api_tool
32 | from . import auth
33 | from .auth import start_callback_server, shutdown_callback_server, auth_manager
34 | from .server import mcp_server
35 | from .utils import logger, META_APP_SECRET
36 | from .pipeboard_auth import pipeboard_auth_manager
37 |
38 | # Only register the login link tool if not explicitly disabled
39 | ENABLE_LOGIN_LINK = not bool(os.environ.get("META_ADS_DISABLE_LOGIN_LINK", ""))
40 |
41 |
42 | async def get_login_link(access_token: Optional[str] = None) -> str:
43 | """
44 | Get a clickable login link for Meta Ads authentication.
45 |
46 | NOTE: This method should only be used if you're using your own Facebook app.
47 | If using Pipeboard authentication (recommended), set the PIPEBOARD_API_TOKEN
48 | environment variable instead (token obtainable via https://pipeboard.co).
49 |
50 | Args:
51 | access_token: Meta API access token (optional - will use cached token if not provided)
52 |
53 | Returns:
54 | A clickable resource link for Meta authentication
55 | """
56 | # Check if we're using pipeboard authentication
57 | using_pipeboard = bool(os.environ.get("PIPEBOARD_API_TOKEN", ""))
58 | callback_server_disabled = bool(os.environ.get("META_ADS_DISABLE_CALLBACK_SERVER", ""))
59 |
60 | if using_pipeboard:
61 | # Pipeboard token-based authentication
62 | try:
63 | logger.info("Using Pipeboard token-based authentication")
64 |
65 | # If an access token was provided, this is likely a test - return success
66 | if access_token:
67 | return json.dumps({
68 | "message": "✅ Authentication Token Provided",
69 | "status": "Using provided access token for authentication",
70 | "token_info": f"Token preview: {access_token[:10]}...",
71 | "authentication_method": "manual_token",
72 | "ready_to_use": "You can now use all Meta Ads MCP tools and commands."
73 | }, indent=2)
74 |
75 | # Check if Pipeboard token is working
76 | token = pipeboard_auth_manager.get_access_token()
77 | if token:
78 | return json.dumps({
79 | "message": "✅ Already Authenticated",
80 | "status": "You're successfully authenticated with Meta Ads via Pipeboard!",
81 | "token_info": f"Token preview: {token[:10]}...",
82 | "authentication_method": "pipeboard_token",
83 | "ready_to_use": "You can now use all Meta Ads MCP tools and commands."
84 | }, indent=2)
85 |
86 | # Start Pipeboard auth flow
87 | auth_data = pipeboard_auth_manager.initiate_auth_flow()
88 | login_url = auth_data.get('loginUrl')
89 |
90 | if login_url:
91 | return json.dumps({
92 | "message": "🔗 Click to Authenticate",
93 | "login_url": login_url,
94 | "markdown_link": f"[🚀 Authenticate with Meta Ads]({login_url})",
95 | "instructions": "Click the link above to complete authentication with Meta Ads.",
96 | "authentication_method": "pipeboard_oauth",
97 | "what_happens_next": "After clicking, you'll be redirected to Meta's authentication page. Once completed, your token will be automatically saved.",
98 | "token_duration": "Your token will be valid for approximately 60 days."
99 | }, indent=2)
100 | else:
101 | return json.dumps({
102 | "message": "❌ Authentication Error",
103 | "error": "Could not generate authentication URL from Pipeboard",
104 | "troubleshooting": [
105 | "Check that your PIPEBOARD_API_TOKEN is valid",
106 | "Ensure the Pipeboard service is accessible",
107 | "Try again in a few moments"
108 | ],
109 | "authentication_method": "pipeboard_oauth_failed"
110 | }, indent=2)
111 |
112 | except Exception as e:
113 | logger.error(f"Error initiating Pipeboard auth flow: {e}")
114 | return json.dumps({
115 | "message": "❌ Pipeboard Authentication Error",
116 | "error": f"Failed to initiate Pipeboard authentication: {str(e)}",
117 | "troubleshooting": [
118 | "✅ Check that PIPEBOARD_API_TOKEN environment variable is set correctly",
119 | "🌐 Verify that pipeboard.co is accessible from your network",
120 | "🔄 Try refreshing your Pipeboard API token",
121 | "⏰ Wait a moment and try again"
122 | ],
123 | "get_help": "Contact support if the issue persists",
124 | "authentication_method": "pipeboard_error"
125 | }, indent=2)
126 | elif callback_server_disabled:
127 | # Production OAuth flow - use Pipeboard OAuth endpoints directly
128 | logger.info("Production OAuth flow - using Pipeboard OAuth endpoints")
129 |
130 | return json.dumps({
131 | "message": "🔐 Authentication Required",
132 | "instructions": "Please sign in to your Pipeboard account to authenticate with Meta Ads.",
133 | "sign_in_url": "https://pipeboard.co/auth/signin",
134 | "markdown_link": "[🚀 Sign in to Pipeboard](https://pipeboard.co/auth/signin)",
135 | "what_to_do": "Click the link above to sign in to your Pipeboard account and complete authentication.",
136 | "authentication_method": "production_oauth"
137 | }, indent=2)
138 | else:
139 | # Original Meta authentication flow (development/local)
140 | # Check if we have a cached token
141 | cached_token = auth_manager.get_access_token()
142 | token_status = "No token" if not cached_token else "Valid token"
143 |
144 | # If we already have a valid token and none was provided, just return success
145 | if cached_token and not access_token:
146 | logger.info("get_login_link called with existing valid token")
147 | return json.dumps({
148 | "message": "✅ Already Authenticated",
149 | "status": "You're successfully authenticated with Meta Ads!",
150 | "token_info": f"Token preview: {cached_token[:10]}...",
151 | "created_at": auth_manager.token_info.created_at if hasattr(auth_manager, "token_info") else None,
152 | "expires_in": auth_manager.token_info.expires_in if hasattr(auth_manager, "token_info") else None,
153 | "authentication_method": "meta_oauth",
154 | "ready_to_use": "You can now use all Meta Ads MCP tools and commands."
155 | }, indent=2)
156 |
157 | # IMPORTANT: Start the callback server first by calling our helper function
158 | # This ensures the server is ready before we provide the URL to the user
159 | logger.info("Starting callback server for authentication")
160 | try:
161 | port = start_callback_server()
162 | logger.info(f"Callback server started on port {port}")
163 |
164 | # Generate direct login URL
165 | auth_manager.redirect_uri = f"http://localhost:{port}/callback" # Ensure port is set correctly
166 | logger.info(f"Setting redirect URI to {auth_manager.redirect_uri}")
167 | login_url = auth_manager.get_auth_url()
168 | logger.info(f"Generated login URL: {login_url}")
169 | except Exception as e:
170 | logger.error(f"Failed to start callback server: {e}")
171 | return json.dumps({
172 | "message": "❌ Local Authentication Unavailable",
173 | "error": "Cannot start local callback server for authentication",
174 | "reason": str(e),
175 | "solutions": [
176 | "🌐 Use Pipeboard authentication: Set PIPEBOARD_API_TOKEN environment variable",
177 | "🔑 Use direct token: Set META_ACCESS_TOKEN environment variable",
178 | "🔧 Check if another service is using the required ports"
179 | ],
180 | "authentication_method": "meta_oauth_disabled"
181 | }, indent=2)
182 |
183 | # Check if we can exchange for long-lived tokens
184 | token_exchange_supported = bool(META_APP_SECRET)
185 | token_duration = "60 days" if token_exchange_supported else "1-2 hours"
186 |
187 | # Return a special format that helps the LLM format the response properly
188 | response = {
189 | "message": "🔗 Click to Authenticate",
190 | "login_url": login_url,
191 | "markdown_link": f"[🚀 Authenticate with Meta Ads]({login_url})",
192 | "instructions": "Click the link above to authenticate with Meta Ads.",
193 | "server_info": f"Local callback server running on port {port}",
194 | "token_duration": f"Your token will be valid for approximately {token_duration}",
195 | "authentication_method": "meta_oauth",
196 | "what_happens_next": "After clicking, you'll be redirected to Meta's authentication page. Once completed, your token will be automatically saved.",
197 | "security_note": "This uses a secure local callback server for development purposes."
198 | }
199 |
200 | # Wait a moment to ensure the server is fully started
201 | await asyncio.sleep(1)
202 |
203 | return json.dumps(response, indent=2)
204 |
205 | # Conditionally register as MCP tool only when enabled
206 | if ENABLE_LOGIN_LINK:
207 | get_login_link = mcp_server.tool()(get_login_link)
```
--------------------------------------------------------------------------------
/tests/test_insights_pagination.py:
--------------------------------------------------------------------------------
```python
1 | """Test pagination functionality for insights endpoint."""
2 |
3 | import pytest
4 | import json
5 | from unittest.mock import AsyncMock, patch
6 | from meta_ads_mcp.core.insights import get_insights
7 |
8 |
9 | class TestInsightsPagination:
10 | """Test suite for pagination functionality in get_insights"""
11 |
12 | @pytest.fixture
13 | def mock_auth_manager(self):
14 | """Mock for the authentication manager"""
15 | with patch('meta_ads_mcp.core.api.auth_manager') as mock, \
16 | patch('meta_ads_mcp.core.auth.get_current_access_token') as mock_get_token:
17 | # Mock a valid access token
18 | mock.get_current_access_token.return_value = "test_access_token"
19 | mock.is_token_valid.return_value = True
20 | mock.app_id = "test_app_id"
21 | mock_get_token.return_value = "test_access_token"
22 | yield mock
23 |
24 | @pytest.fixture
25 | def valid_account_id(self):
26 | return "act_701351919139047"
27 |
28 | @pytest.fixture
29 | def mock_paginated_response_page1(self):
30 | """Mock first page of paginated response"""
31 | return {
32 | "data": [
33 | {
34 | "campaign_id": "campaign_1",
35 | "campaign_name": "Test Campaign 1",
36 | "spend": "100.50",
37 | "impressions": "1000",
38 | "clicks": "50"
39 | },
40 | {
41 | "campaign_id": "campaign_2",
42 | "campaign_name": "Test Campaign 2",
43 | "spend": "200.75",
44 | "impressions": "2000",
45 | "clicks": "100"
46 | }
47 | ],
48 | "paging": {
49 | "cursors": {
50 | "before": "before_cursor_1",
51 | "after": "after_cursor_1"
52 | },
53 | "next": "https://graph.facebook.com/v20.0/act_123/insights?after=after_cursor_1&limit=2"
54 | }
55 | }
56 |
57 | @pytest.fixture
58 | def mock_paginated_response_page2(self):
59 | """Mock second page of paginated response"""
60 | return {
61 | "data": [
62 | {
63 | "campaign_id": "campaign_3",
64 | "campaign_name": "Test Campaign 3",
65 | "spend": "150.25",
66 | "impressions": "1500",
67 | "clicks": "75"
68 | }
69 | ],
70 | "paging": {
71 | "cursors": {
72 | "before": "before_cursor_2",
73 | "after": "after_cursor_2"
74 | }
75 | }
76 | }
77 |
78 | @pytest.mark.asyncio
79 | async def test_insights_with_limit_parameter(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
80 | """Test that limit parameter is properly passed to API"""
81 | with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
82 | mock_api_request.return_value = mock_paginated_response_page1
83 |
84 | result = await get_insights(
85 | object_id=valid_account_id,
86 | level="campaign",
87 | time_range="last_30d",
88 | limit=2
89 | )
90 |
91 | # Verify the API was called with correct parameters
92 | mock_api_request.assert_called_once()
93 | call_args = mock_api_request.call_args
94 |
95 | # Check that limit is included in params
96 | params = call_args[0][2]
97 | assert params["limit"] == 2
98 | assert params["level"] == "campaign"
99 | assert params["date_preset"] == "last_30d"
100 |
101 | # Verify the response structure
102 | result_data = json.loads(result)
103 | assert "data" in result_data
104 | assert len(result_data["data"]) == 2
105 | assert "paging" in result_data
106 |
107 | @pytest.mark.asyncio
108 | async def test_insights_with_after_cursor(self, mock_auth_manager, valid_account_id, mock_paginated_response_page2):
109 | """Test that after cursor is properly passed to API for pagination"""
110 | with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
111 | mock_api_request.return_value = mock_paginated_response_page2
112 |
113 | after_cursor = "after_cursor_1"
114 | result = await get_insights(
115 | object_id=valid_account_id,
116 | level="campaign",
117 | time_range="last_30d",
118 | limit=10,
119 | after=after_cursor
120 | )
121 |
122 | # Verify the API was called with correct parameters
123 | mock_api_request.assert_called_once()
124 | call_args = mock_api_request.call_args
125 |
126 | # Check that after cursor is included in params
127 | params = call_args[0][2]
128 | assert params["after"] == after_cursor
129 | assert params["limit"] == 10
130 |
131 | # Verify the response structure
132 | result_data = json.loads(result)
133 | assert "data" in result_data
134 | assert len(result_data["data"]) == 1
135 |
136 | @pytest.mark.asyncio
137 | async def test_insights_default_limit(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
138 | """Test that default limit is 25 when not specified"""
139 | with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
140 | mock_api_request.return_value = mock_paginated_response_page1
141 |
142 | result = await get_insights(
143 | object_id=valid_account_id,
144 | level="campaign"
145 | )
146 |
147 | # Verify the API was called with default limit
148 | mock_api_request.assert_called_once()
149 | call_args = mock_api_request.call_args
150 |
151 | params = call_args[0][2]
152 | assert params["limit"] == 25 # Default value
153 |
154 | @pytest.mark.asyncio
155 | async def test_insights_without_after_cursor(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
156 | """Test that after parameter is not included when empty"""
157 | with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
158 | mock_api_request.return_value = mock_paginated_response_page1
159 |
160 | result = await get_insights(
161 | object_id=valid_account_id,
162 | level="campaign",
163 | after="" # Empty after cursor
164 | )
165 |
166 | # Verify the API was called without after parameter
167 | mock_api_request.assert_called_once()
168 | call_args = mock_api_request.call_args
169 |
170 | params = call_args[0][2]
171 | assert "after" not in params # Should not be included when empty
172 |
173 | @pytest.mark.asyncio
174 | async def test_insights_pagination_with_custom_time_range(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
175 | """Test pagination works with custom time range"""
176 | with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
177 | mock_api_request.return_value = mock_paginated_response_page1
178 |
179 | custom_time_range = {"since": "2024-01-01", "until": "2024-01-31"}
180 | result = await get_insights(
181 | object_id=valid_account_id,
182 | level="campaign",
183 | time_range=custom_time_range,
184 | limit=5,
185 | after="test_cursor"
186 | )
187 |
188 | # Verify the API was called with correct parameters
189 | mock_api_request.assert_called_once()
190 | call_args = mock_api_request.call_args
191 |
192 | params = call_args[0][2]
193 | assert params["limit"] == 5
194 | assert params["after"] == "test_cursor"
195 | assert params["time_range"] == json.dumps(custom_time_range)
196 |
197 | @pytest.mark.asyncio
198 | async def test_insights_pagination_with_breakdown(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
199 | """Test pagination works with breakdown parameter"""
200 | with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
201 | mock_api_request.return_value = mock_paginated_response_page1
202 |
203 | result = await get_insights(
204 | object_id=valid_account_id,
205 | level="campaign",
206 | breakdown="age",
207 | limit=10,
208 | after="test_cursor_2"
209 | )
210 |
211 | # Verify the API was called with correct parameters
212 | mock_api_request.assert_called_once()
213 | call_args = mock_api_request.call_args
214 |
215 | params = call_args[0][2]
216 | assert params["limit"] == 10
217 | assert params["after"] == "test_cursor_2"
218 | assert params["breakdowns"] == "age"
219 |
220 | @pytest.mark.asyncio
221 | async def test_insights_large_limit_value(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
222 | """Test that large limit values are accepted (API will enforce its own limits)"""
223 | with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
224 | mock_api_request.return_value = mock_paginated_response_page1
225 |
226 | result = await get_insights(
227 | object_id=valid_account_id,
228 | level="campaign",
229 | limit=1000 # Large limit - API will enforce its own max
230 | )
231 |
232 | # Verify the API was called with the large limit
233 | mock_api_request.assert_called_once()
234 | call_args = mock_api_request.call_args
235 |
236 | params = call_args[0][2]
237 | assert params["limit"] == 1000
238 |
239 | @pytest.mark.asyncio
240 | async def test_insights_paging_response_structure(self, mock_auth_manager, valid_account_id, mock_paginated_response_page1):
241 | """Test that paging information is preserved in the response"""
242 | with patch('meta_ads_mcp.core.insights.make_api_request', new_callable=AsyncMock) as mock_api_request:
243 | mock_api_request.return_value = mock_paginated_response_page1
244 |
245 | result = await get_insights(
246 | object_id=valid_account_id,
247 | level="campaign",
248 | limit=2
249 | )
250 |
251 | # Verify the response includes paging information
252 | result_data = json.loads(result)
253 | assert "data" in result_data
254 | assert "paging" in result_data
255 | assert "cursors" in result_data["paging"]
256 | assert "after" in result_data["paging"]["cursors"]
257 | assert "next" in result_data["paging"]
258 |
```
--------------------------------------------------------------------------------
/tests/test_duplication.py:
--------------------------------------------------------------------------------
```python
1 | """Tests for the duplication module."""
2 |
3 | import os
4 | import json
5 | import pytest
6 | from unittest.mock import patch, AsyncMock, Mock
7 | from meta_ads_mcp.core.duplication import ENABLE_DUPLICATION
8 |
9 |
10 | def test_duplication_disabled_by_default():
11 | """Test that duplication is disabled by default."""
12 | # Test with no environment variable set
13 | with patch.dict(os.environ, {}, clear=True):
14 | from meta_ads_mcp.core import duplication
15 | # When imported fresh, it should be disabled
16 | assert not duplication.ENABLE_DUPLICATION
17 |
18 |
19 | def test_duplication_enabled_with_env_var():
20 | """Test that duplication is enabled when environment variable is set."""
21 | with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
22 | # Need to reload the module to pick up the new environment variable
23 | import importlib
24 | from meta_ads_mcp.core import duplication
25 | importlib.reload(duplication)
26 | assert duplication.ENABLE_DUPLICATION
27 |
28 |
29 | @pytest.mark.asyncio
30 | async def test_forward_duplication_request_no_pipeboard_token():
31 | """Test that _forward_duplication_request handles missing Pipeboard token."""
32 | from meta_ads_mcp.core.duplication import _forward_duplication_request
33 |
34 | # Mock the auth integration to return no Pipeboard token but a Facebook token
35 | with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
36 | mock_auth.get_pipeboard_token.return_value = None # No Pipeboard token
37 | mock_auth.get_auth_token.return_value = "facebook_token" # Has Facebook token
38 |
39 | result = await _forward_duplication_request("campaign", "123456789", None, {})
40 | result_json = json.loads(result)
41 |
42 | assert result_json["error"] == "authentication_required"
43 | assert "Pipeboard API token not found" in result_json["message"]
44 |
45 |
46 | @pytest.mark.asyncio
47 | async def test_forward_duplication_request_no_facebook_token():
48 | """Test that _forward_duplication_request handles missing Facebook token."""
49 | from meta_ads_mcp.core.duplication import _forward_duplication_request
50 |
51 | # Mock the auth integration to return Pipeboard token but no Facebook token
52 | with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
53 | mock_auth.get_pipeboard_token.return_value = "pipeboard_token" # Has Pipeboard token
54 | mock_auth.get_auth_token.return_value = None # No Facebook token
55 |
56 | # Mock get_current_access_token to also return None
57 | with patch("meta_ads_mcp.core.auth.get_current_access_token") as mock_get_token:
58 | mock_get_token.return_value = None
59 |
60 | result = await _forward_duplication_request("campaign", "123456789", None, {})
61 | result_json = json.loads(result)
62 |
63 | assert result_json["error"] == "authentication_required"
64 | assert "Meta Ads access token not found" in result_json["message"]
65 |
66 |
67 | @pytest.mark.asyncio
68 | async def test_forward_duplication_request_with_both_tokens():
69 | """Test that _forward_duplication_request makes HTTP request with dual headers."""
70 | from meta_ads_mcp.core.duplication import _forward_duplication_request
71 |
72 | mock_response = Mock()
73 | mock_response.status_code = 403
74 | mock_response.json.return_value = {"error": "premium_feature"}
75 |
76 | # Mock the auth integration to return both tokens
77 | with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
78 | mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
79 | mock_auth.get_auth_token.return_value = "facebook_token"
80 |
81 | with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
82 | mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
83 |
84 | result = await _forward_duplication_request("campaign", "123456789", None, {
85 | "name_suffix": " - Test"
86 | })
87 | result_json = json.loads(result)
88 |
89 | # Should return premium feature message for 403 response
90 | assert result_json["error"] == "premium_feature_required"
91 | assert "premium feature" in result_json["message"]
92 |
93 | # Verify the HTTP request was made with correct parameters
94 | mock_client.return_value.__aenter__.return_value.post.assert_called_once()
95 | call_args = mock_client.return_value.__aenter__.return_value.post.call_args
96 |
97 | # Check URL
98 | assert call_args[0][0] == "https://mcp.pipeboard.co/api/meta/duplicate/campaign/123456789"
99 |
100 | # Check dual headers (the key change!)
101 | headers = call_args[1]["headers"]
102 | assert headers["Authorization"] == "Bearer facebook_token" # Facebook token for Meta API
103 | assert headers["X-Pipeboard-Token"] == "pipeboard_token" # Pipeboard token for auth
104 | assert headers["Content-Type"] == "application/json"
105 |
106 | # Check JSON payload
107 | json_payload = call_args[1]["json"]
108 | assert json_payload == {"name_suffix": " - Test"}
109 |
110 |
111 | @pytest.mark.asyncio
112 | async def test_forward_duplication_request_with_provided_access_token():
113 | """Test that provided access_token parameter is used when available."""
114 | from meta_ads_mcp.core.duplication import _forward_duplication_request
115 |
116 | mock_response = Mock()
117 | mock_response.status_code = 200
118 | mock_response.json.return_value = {"success": True, "new_campaign_id": "987654321"}
119 |
120 | # Mock the auth integration to return Pipeboard token but no Facebook token in context
121 | with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
122 | mock_auth.get_pipeboard_token.return_value = "pipeboard_token"
123 | mock_auth.get_auth_token.return_value = None # No Facebook token in context
124 |
125 | with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
126 | mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
127 |
128 | # Provide access_token as parameter
129 | result = await _forward_duplication_request("campaign", "123456789", "provided_facebook_token", {})
130 | result_json = json.loads(result)
131 |
132 | # Should succeed
133 | assert result_json["success"] is True
134 |
135 | # Verify the HTTP request used the provided token
136 | call_args = mock_client.return_value.__aenter__.return_value.post.call_args
137 | headers = call_args[1]["headers"]
138 | assert headers["Authorization"] == "Bearer provided_facebook_token"
139 | assert headers["X-Pipeboard-Token"] == "pipeboard_token"
140 |
141 |
142 | @pytest.mark.asyncio
143 | async def test_duplicate_campaign_function_available_when_enabled():
144 | """Test that duplicate_campaign function is available when feature is enabled."""
145 | with patch.dict(os.environ, {"META_ADS_ENABLE_DUPLICATION": "1"}):
146 | # Reload module to pick up environment variable
147 | import importlib
148 | from meta_ads_mcp.core import duplication
149 | importlib.reload(duplication)
150 |
151 | # Function should be available
152 | assert hasattr(duplication, 'duplicate_campaign')
153 |
154 | # Test that it calls the forwarding function
155 | with patch("meta_ads_mcp.core.duplication._forward_duplication_request") as mock_forward:
156 | mock_forward.return_value = '{"success": true}'
157 |
158 | result = await duplication.duplicate_campaign("123456789", access_token="test_token")
159 |
160 | mock_forward.assert_called_once_with(
161 | "campaign",
162 | "123456789",
163 | "test_token",
164 | {
165 | "name_suffix": " - Copy",
166 | "include_ad_sets": True,
167 | "include_ads": True,
168 | "include_creatives": True,
169 | "copy_schedule": False,
170 | "new_daily_budget": None,
171 | "new_status": "PAUSED"
172 | }
173 | )
174 |
175 |
176 | def test_get_estimated_components():
177 | """Test the _get_estimated_components helper function."""
178 | from meta_ads_mcp.core.duplication import _get_estimated_components
179 |
180 | # Test campaign with all components
181 | campaign_result = _get_estimated_components("campaign", {
182 | "include_ad_sets": True,
183 | "include_ads": True,
184 | "include_creatives": True
185 | })
186 | assert campaign_result["campaigns"] == 1
187 | assert "ad_sets" in campaign_result
188 | assert "ads" in campaign_result
189 | assert "creatives" in campaign_result
190 |
191 | # Test adset
192 | adset_result = _get_estimated_components("adset", {"include_ads": True})
193 | assert adset_result["ad_sets"] == 1
194 | assert "ads" in adset_result
195 |
196 | # Test creative only
197 | creative_result = _get_estimated_components("creative", {})
198 | assert creative_result == {"creatives": 1}
199 |
200 |
201 | @pytest.mark.asyncio
202 | async def test_dual_header_authentication_integration():
203 | """Test that the dual-header authentication works end-to-end."""
204 | from meta_ads_mcp.core.duplication import _forward_duplication_request
205 |
206 | mock_response = Mock()
207 | mock_response.status_code = 200
208 | mock_response.json.return_value = {
209 | "success": True,
210 | "new_campaign_id": "987654321",
211 | "subscription": {"status": "active"}
212 | }
213 |
214 | # Test the complete dual-header flow
215 | with patch("meta_ads_mcp.core.duplication.FastMCPAuthIntegration") as mock_auth:
216 | mock_auth.get_pipeboard_token.return_value = "pb_token_12345"
217 | mock_auth.get_auth_token.return_value = "fb_token_67890"
218 |
219 | with patch("meta_ads_mcp.core.duplication.httpx.AsyncClient") as mock_client:
220 | mock_client.return_value.__aenter__.return_value.post.return_value = mock_response
221 |
222 | result = await _forward_duplication_request("adset", "456789", None, {
223 | "target_campaign_id": "123456",
224 | "include_ads": True
225 | })
226 | result_json = json.loads(result)
227 |
228 | # Should succeed
229 | assert result_json["success"] is True
230 | assert result_json["new_campaign_id"] == "987654321"
231 |
232 | # Verify correct endpoint was called
233 | call_args = mock_client.return_value.__aenter__.return_value.post.call_args
234 | assert "adset/456789" in call_args[0][0]
235 |
236 | # Verify dual headers were sent correctly
237 | headers = call_args[1]["headers"]
238 | assert headers["Authorization"] == "Bearer fb_token_67890"
239 | assert headers["X-Pipeboard-Token"] == "pb_token_12345"
240 |
241 | # Verify payload
242 | payload = call_args[1]["json"]
243 | assert payload["target_campaign_id"] == "123456"
244 | assert payload["include_ads"] is True
```
--------------------------------------------------------------------------------
/LOCAL_INSTALLATION.md:
--------------------------------------------------------------------------------
```markdown
1 | # Meta Ads MCP - Local Installation Guide
2 |
3 | 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)**.
4 |
5 | ## Table of Contents
6 |
7 | - [Prerequisites](#prerequisites)
8 | - [Installation Methods](#installation-methods)
9 | - [Authentication Setup](#authentication-setup)
10 | - [MCP Client Configuration](#mcp-client-configuration)
11 | - [Development Installation](#development-installation)
12 | - [Privacy and Security](#privacy-and-security)
13 | - [Testing and Verification](#testing-and-verification)
14 | - [Debugging and Logs](#debugging-and-logs)
15 | - [Troubleshooting](#troubleshooting)
16 | - [Advanced Configuration](#advanced-configuration)
17 |
18 | ## Prerequisites
19 |
20 | - **Python 3.8 or higher**
21 | - **[uv](https://docs.astral.sh/uv/) package manager** (recommended) or pip
22 | - **Meta Ads account** with appropriate permissions
23 | - **MCP-compatible client** (Claude Desktop, Cursor, Cherry Studio, etc.)
24 |
25 | ## Installation Methods
26 |
27 | ### Method 1: Using uvx (Recommended)
28 |
29 | ```bash
30 | # Install via uvx (automatically handles dependencies)
31 | uvx meta-ads-mcp
32 | ```
33 |
34 | ### Method 2: Using pip
35 |
36 | ```bash
37 | # Install via pip
38 | pip install meta-ads-mcp
39 | ```
40 |
41 | ### Method 3: Development Installation
42 |
43 | ```bash
44 | # Clone the repository
45 | git clone https://github.com/pipeboard-co/meta-ads-mcp.git
46 | cd meta-ads-mcp
47 |
48 | # Install in development mode
49 | uv pip install -e .
50 | # Or with pip
51 | pip install -e .
52 | ```
53 |
54 | ## Authentication Setup
55 |
56 | You have two authentication options:
57 |
58 | ### Option 1: Pipeboard Authentication (Recommended)
59 |
60 | This is the easiest method that handles all OAuth complexity for you:
61 |
62 | 1. **Sign up to Pipeboard**
63 | - Visit [Pipeboard.co](https://pipeboard.co)
64 | - Create an account
65 |
66 | 2. **Generate API Token**
67 | - Go to [pipeboard.co/api-tokens](https://pipeboard.co/api-tokens)
68 | - Generate a new API token
69 | - Copy the token securely
70 |
71 | 3. **Set Environment Variable**
72 | ```bash
73 | # On macOS/Linux
74 | export PIPEBOARD_API_TOKEN=your_pipeboard_token_here
75 |
76 | # On Windows (Command Prompt)
77 | set PIPEBOARD_API_TOKEN=your_pipeboard_token_here
78 |
79 | # On Windows (PowerShell)
80 | $env:PIPEBOARD_API_TOKEN="your_pipeboard_token_here"
81 | ```
82 |
83 | 4. **Make it Persistent**
84 |
85 | Add to your shell profile (`.bashrc`, `.zshrc`, etc.):
86 | ```bash
87 | echo 'export PIPEBOARD_API_TOKEN=your_pipeboard_token_here' >> ~/.bashrc
88 | source ~/.bashrc
89 | ```
90 |
91 | ### Option 2: Custom Meta App
92 |
93 | If you prefer to use your own Meta Developer App, see [CUSTOM_META_APP.md](CUSTOM_META_APP.md) for detailed instructions.
94 |
95 | ## MCP Client Configuration
96 |
97 | ### Claude Desktop
98 |
99 | Add to your `claude_desktop_config.json`:
100 |
101 | **Location:**
102 | - **macOS**: `~/Library/Application Support/Claude/claude_desktop_config.json`
103 | - **Windows**: `%APPDATA%\Claude\claude_desktop_config.json`
104 |
105 | **Configuration:**
106 | ```json
107 | {
108 | "mcpServers": {
109 | "meta-ads": {
110 | "command": "uvx",
111 | "args": ["meta-ads-mcp"],
112 | "env": {
113 | "PIPEBOARD_API_TOKEN": "your_pipeboard_token"
114 | }
115 | }
116 | }
117 | }
118 | ```
119 |
120 | ### Cursor
121 |
122 | Add to your `~/.cursor/mcp.json`:
123 |
124 | ```json
125 | {
126 | "mcpServers": {
127 | "meta-ads": {
128 | "command": "uvx",
129 | "args": ["meta-ads-mcp"],
130 | "env": {
131 | "PIPEBOARD_API_TOKEN": "your_pipeboard_token"
132 | }
133 | }
134 | }
135 | }
136 | ```
137 |
138 | ### Cherry Studio
139 |
140 | In Cherry Studio settings, add a new MCP server:
141 | - **Name**: Meta Ads MCP
142 | - **Command**: `uvx`
143 | - **Arguments**: `["meta-ads-mcp"]`
144 | - **Environment Variables**: `PIPEBOARD_API_TOKEN=your_pipeboard_token`
145 |
146 | ## Development Installation
147 |
148 | ### Setting Up Development Environment
149 |
150 | ```bash
151 | # Clone the repository
152 | git clone https://github.com/pipeboard-co/meta-ads-mcp.git
153 | cd meta-ads-mcp
154 |
155 | # Create virtual environment (optional but recommended)
156 | python -m venv venv
157 | source venv/bin/activate # On Windows: venv\Scripts\activate
158 |
159 | # Install in development mode with dependencies
160 | uv pip install -e .
161 |
162 | # Install development dependencies
163 | uv pip install -e ".[dev]" # If dev dependencies are defined
164 | ```
165 |
166 | ### Running from Source
167 |
168 | ```bash
169 | # Set environment variable
170 | export PIPEBOARD_API_TOKEN=your_token
171 |
172 | # Run directly
173 | python -m meta_ads_mcp
174 |
175 | # Or if installed in development mode
176 | meta-ads-mcp
177 | ```
178 |
179 | ### Testing Your Installation
180 |
181 | ```bash
182 | # Test the installation
183 | python -c "import meta_ads_mcp; print('Installation successful!')"
184 |
185 | # Test MCP server startup
186 | meta-ads-mcp --help
187 | ```
188 |
189 | ## Privacy and Security
190 |
191 | ### Token Storage and Caching
192 |
193 | Meta Ads MCP follows security best practices:
194 |
195 | 1. **Secure Token Cache Location**:
196 | - **Windows**: `%APPDATA%\meta-ads-mcp\token_cache.json`
197 | - **macOS**: `~/Library/Application Support/meta-ads-mcp/token_cache.json`
198 | - **Linux**: `~/.config/meta-ads-mcp/token_cache.json`
199 |
200 | 2. **Automatic Token Management**:
201 | - Tokens are cached securely after first authentication
202 | - You don't need to provide access tokens for each command
203 | - Tokens are automatically refreshed when needed
204 |
205 | 3. **Environment Variable Security**:
206 | - `PIPEBOARD_API_TOKEN` should be kept secure
207 | - Don't commit tokens to version control
208 | - Use environment files (`.env`) for local development
209 |
210 | ### Security Best Practices
211 |
212 | ```bash
213 | # Create a .env file for local development (never commit this)
214 | echo "PIPEBOARD_API_TOKEN=your_token_here" > .env
215 |
216 | # Add .env to .gitignore
217 | echo ".env" >> .gitignore
218 |
219 | # Load environment variables from .env
220 | source .env
221 | ```
222 |
223 | ## Testing and Verification
224 |
225 | ### Basic Functionality Test
226 |
227 | Once installed and configured, test with your MCP client:
228 |
229 | 1. **Verify Account Access**
230 | ```
231 | Ask your LLM: "Use mcp_meta_ads_get_ad_accounts to show my Meta ad accounts"
232 | ```
233 |
234 | 2. **Check Account Details**
235 | ```
236 | Ask your LLM: "Get details for account act_XXXXXXXXX using mcp_meta_ads_get_account_info"
237 | ```
238 |
239 | 3. **List Campaigns**
240 | ```
241 | Ask your LLM: "Show me my active campaigns using mcp_meta_ads_get_campaigns"
242 | ```
243 |
244 | ### Manual Testing with Python
245 |
246 | ```python
247 | # Test authentication
248 | from meta_ads_mcp.core.auth import get_access_token
249 |
250 | try:
251 | token = get_access_token()
252 | print("Authentication successful!")
253 | print(f"Token starts with: {token[:10]}...")
254 | except Exception as e:
255 | print(f"Authentication failed: {e}")
256 | ```
257 |
258 | ### Testing with MCP Client
259 |
260 | When using Meta Ads MCP with an LLM interface:
261 |
262 | 1. Ensure the `PIPEBOARD_API_TOKEN` environment variable is set
263 | 2. Verify account access by calling `mcp_meta_ads_get_ad_accounts`
264 | 3. Check specific account details with `mcp_meta_ads_get_account_info`
265 | 4. Test campaign retrieval with `mcp_meta_ads_get_campaigns`
266 |
267 | ## Debugging and Logs
268 |
269 | ### Log File Locations
270 |
271 | Debug logs are automatically created in platform-specific locations:
272 |
273 | - **macOS**: `~/Library/Application\ Support/meta-ads-mcp/meta_ads_debug.log`
274 | - **Windows**: `%APPDATA%\meta-ads-mcp\meta_ads_debug.log`
275 | - **Linux**: `~/.config/meta-ads-mcp/meta_ads_debug.log`
276 |
277 | ### Enabling Debug Mode
278 |
279 | ```bash
280 | # Set debug environment variable
281 | export META_ADS_DEBUG=true
282 |
283 | # Run with verbose output
284 | meta-ads-mcp --verbose
285 | ```
286 |
287 | ### Viewing Logs
288 |
289 | ```bash
290 | # On macOS/Linux
291 | tail -f ~/Library/Application\ Support/meta-ads-mcp/meta_ads_debug.log
292 |
293 | # On Windows
294 | type %APPDATA%\meta-ads-mcp\meta_ads_debug.log
295 | ```
296 |
297 | ### Common Debug Commands
298 |
299 | ```bash
300 | # Check if MCP server starts correctly
301 | meta-ads-mcp --test-connection
302 |
303 | # Verify environment variables
304 | echo $PIPEBOARD_API_TOKEN
305 |
306 | # Test Pipeboard authentication
307 | python -c "
308 | from meta_ads_mcp.core.pipeboard_auth import test_auth
309 | test_auth()
310 | "
311 | ```
312 |
313 | ## Troubleshooting
314 |
315 | ### Authentication Issues
316 |
317 | #### Problem: "PIPEBOARD_API_TOKEN not set"
318 | ```bash
319 | # Solution: Set the environment variable
320 | export PIPEBOARD_API_TOKEN=your_token_here
321 |
322 | # Verify it's set
323 | echo $PIPEBOARD_API_TOKEN
324 | ```
325 |
326 | #### Problem: "Invalid Pipeboard token"
327 | 1. Check your token at [pipeboard.co/api-tokens](https://pipeboard.co/api-tokens)
328 | 2. Regenerate if necessary
329 | 3. Update your environment variable
330 |
331 | #### Problem: "Authentication failed"
332 | ```bash
333 | # Clear cached tokens and retry
334 | rm -rf ~/.config/meta-ads-mcp/token_cache.json # Linux
335 | rm -rf ~/Library/Application\ Support/meta-ads-mcp/token_cache.json # macOS
336 |
337 | # Force re-authentication
338 | python test_pipeboard_auth.py --force-login
339 | ```
340 |
341 | ### Installation Issues
342 |
343 | #### Problem: "Command not found: uvx"
344 | ```bash
345 | # Install uv first
346 | curl -LsSf https://astral.sh/uv/install.sh | sh
347 |
348 | # Or use pip
349 | pip install meta-ads-mcp
350 | ```
351 |
352 | #### Problem: "Permission denied"
353 | ```bash
354 | # Use user installation
355 | pip install --user meta-ads-mcp
356 |
357 | # Or use virtual environment
358 | python -m venv venv
359 | source venv/bin/activate
360 | pip install meta-ads-mcp
361 | ```
362 |
363 | #### Problem: "Python version incompatible"
364 | ```bash
365 | # Check Python version
366 | python --version
367 |
368 | # Update to Python 3.8+
369 | # Use pyenv or your system's package manager
370 | ```
371 |
372 | ### Runtime Issues
373 |
374 | #### Problem: "Failed to connect to Meta API"
375 | 1. Check internet connection
376 | 2. Verify Meta API status
377 | 3. Check rate limits
378 | 4. Ensure account permissions
379 |
380 | #### Problem: "MCP client can't find server"
381 | 1. Verify the command path in your MCP client config
382 | 2. Check environment variables are set in the client
383 | 3. Test the command manually in terminal
384 |
385 | #### Problem: "SSL/TLS errors"
386 | ```bash
387 | # Update certificates
388 | pip install --upgrade certifi
389 |
390 | # Or use system certificates
391 | export SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt
392 | ```
393 |
394 | ### API Errors
395 |
396 | #### Problem: "Insufficient permissions"
397 | - Ensure your Meta account has access to the ad accounts
398 | - Check if your Pipeboard token has the right scopes
399 | - Verify account roles in Meta Business Manager
400 |
401 | #### Problem: "Rate limit exceeded"
402 | - Wait before retrying
403 | - Reduce request frequency
404 | - Check if multiple instances are running
405 |
406 | #### Problem: "Account not found"
407 | - Verify account ID format (should be `act_XXXXXXXXX`)
408 | - Check account access permissions
409 | - Ensure account is active
410 |
411 | ### Performance Issues
412 |
413 | #### Problem: "Slow response times"
414 | ```bash
415 | # Check network latency
416 | ping graph.facebook.com
417 |
418 | # Clear cache
419 | rm -rf ~/.config/meta-ads-mcp/token_cache.json
420 |
421 | # Check system resources
422 | top # or htop on Linux/macOS
423 | ```
424 |
425 | ## Advanced Configuration
426 |
427 | ### Custom Configuration File
428 |
429 | Create `~/.config/meta-ads-mcp/config.json`:
430 |
431 | ```json
432 | {
433 | "api_version": "v21.0",
434 | "timeout": 30,
435 | "max_retries": 3,
436 | "debug": false,
437 | "cache_duration": 3600
438 | }
439 | ```
440 |
441 | ### Environment Variables
442 |
443 | ```bash
444 | # API Configuration
445 | export META_API_VERSION=v21.0
446 | export META_API_TIMEOUT=30
447 | export META_ADS_DEBUG=true
448 |
449 | # Cache Configuration
450 | export META_ADS_CACHE_DIR=/custom/cache/path
451 | export META_ADS_CACHE_DURATION=3600
452 |
453 | # Pipeboard Configuration
454 | export PIPEBOARD_API_BASE=https://api.pipeboard.co
455 | export PIPEBOARD_API_TOKEN=your_token_here
456 | ```
457 |
458 | ### Transport Configuration
459 |
460 | Meta Ads MCP uses **stdio transport** by default. For HTTP transport:
461 |
462 | See [STREAMABLE_HTTP_SETUP.md](STREAMABLE_HTTP_SETUP.md) for streamable HTTP transport configuration.
463 |
464 | ### Custom Meta App Integration
465 |
466 | For advanced users who want to use their own Meta Developer App:
467 |
468 | 1. Follow [CUSTOM_META_APP.md](CUSTOM_META_APP.md) guide
469 | 2. Set up OAuth flow
470 | 3. Configure environment variables:
471 | ```bash
472 | export META_APP_ID=your_app_id
473 | export META_APP_SECRET=your_app_secret
474 | export META_REDIRECT_URI=your_redirect_uri
475 | ```
476 |
477 | ## Getting Help
478 |
479 | If you're still experiencing issues:
480 |
481 | 1. **Check the logs** for detailed error messages
482 | 2. **Search existing issues** on GitHub
483 | 3. **Join our Discord** at [discord.gg/YzMwQ8zrjr](https://discord.gg/YzMwQ8zrjr)
484 | 4. **Email support** at [email protected]
485 | 5. **Consider Remote MCP** at [pipeboard.co](https://pipeboard.co) as an alternative
486 |
487 | ---
488 |
489 | **Quick Alternative**: If local installation is causing issues, try our [Remote MCP service](https://pipeboard.co) - no local setup required!
490 |
```
--------------------------------------------------------------------------------
/tests/test_account_search.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Focused Account Search Test for Meta Ads MCP
4 |
5 | This test validates that the search tool correctly finds and returns
6 | account data for known test accounts.
7 |
8 | Expected test accounts:
9 | - act_4891437610982483 (Yves Junqueira)
10 | - act_701351919139047 (Injury Payouts)
11 | """
12 |
13 | import requests
14 | import json
15 | import os
16 | import sys
17 | from typing import Dict, Any, List
18 |
19 | # Load environment variables from .env file
20 | try:
21 | from dotenv import load_dotenv
22 | load_dotenv()
23 | print("✅ Loaded environment variables from .env file")
24 | except ImportError:
25 | print("⚠️ python-dotenv not installed, using system environment variables only")
26 |
27 | class AccountSearchTester:
28 | """Test suite focused on account search functionality"""
29 |
30 | def __init__(self, base_url: str = "http://localhost:8080"):
31 | self.base_url = base_url.rstrip('/')
32 | self.endpoint = f"{self.base_url}/mcp/"
33 | self.request_id = 1
34 |
35 | # Expected test data
36 | self.expected_accounts = [
37 | {
38 | "id": "act_4891437610982483",
39 | "name": "Yves Junqueira",
40 | "account_id": "4891437610982483"
41 | },
42 | {
43 | "id": "act_701351919139047",
44 | "name": "Injury Payouts",
45 | "account_id": "701351919139047"
46 | }
47 | ]
48 |
49 | def _make_request(self, method: str, params: Dict[str, Any] = None,
50 | headers: Dict[str, str] = None) -> Dict[str, Any]:
51 | """Make a JSON-RPC request to the MCP server"""
52 |
53 | default_headers = {
54 | "Content-Type": "application/json",
55 | "Accept": "application/json, text/event-stream",
56 | "User-Agent": "Account-Search-Test-Client/1.0"
57 | }
58 |
59 | if headers:
60 | default_headers.update(headers)
61 |
62 | payload = {
63 | "jsonrpc": "2.0",
64 | "method": method,
65 | "id": self.request_id
66 | }
67 |
68 | if params:
69 | payload["params"] = params
70 |
71 | try:
72 | response = requests.post(
73 | self.endpoint,
74 | headers=default_headers,
75 | json=payload,
76 | timeout=10
77 | )
78 |
79 | self.request_id += 1
80 |
81 | return {
82 | "status_code": response.status_code,
83 | "headers": dict(response.headers),
84 | "json": response.json() if response.status_code == 200 else None,
85 | "text": response.text,
86 | "success": response.status_code == 200
87 | }
88 |
89 | except requests.exceptions.RequestException as e:
90 | return {
91 | "status_code": 0,
92 | "headers": {},
93 | "json": None,
94 | "text": str(e),
95 | "success": False,
96 | "error": str(e)
97 | }
98 |
99 | def test_search_accounts(self) -> Dict[str, Any]:
100 | """Test searching for accounts with various queries"""
101 |
102 | queries_to_test = [
103 | "accounts",
104 | "ad accounts",
105 | "meta accounts",
106 | "Yves",
107 | "Injury Payouts"
108 | ]
109 |
110 | results = {}
111 |
112 | for query in queries_to_test:
113 | print(f"\n🔍 Testing search query: '{query}'")
114 |
115 | result = self._make_request("tools/call", {
116 | "name": "search",
117 | "arguments": {"query": query}
118 | })
119 |
120 | if not result["success"]:
121 | results[query] = {
122 | "success": False,
123 | "error": result.get("text", "Unknown error")
124 | }
125 | print(f"❌ Search failed: {result.get('text', 'Unknown error')}")
126 | continue
127 |
128 | # Parse the tool response
129 | response_data = result["json"]["result"]
130 | content = response_data.get("content", [{}])[0].get("text", "")
131 |
132 | try:
133 | parsed_content = json.loads(content)
134 | ids = parsed_content.get("ids", [])
135 |
136 | # Check for expected account IDs
137 | expected_account_ids = [f"account:{acc['id']}" for acc in self.expected_accounts]
138 | found_account_ids = [id for id in ids if id.startswith("account:")]
139 |
140 | results[query] = {
141 | "success": True,
142 | "ids": ids,
143 | "account_ids": found_account_ids,
144 | "found_expected_accounts": len([id for id in expected_account_ids if id in found_account_ids]),
145 | "total_expected": len(expected_account_ids),
146 | "raw_content": parsed_content
147 | }
148 |
149 | print(f"✅ Found {len(found_account_ids)} account IDs: {found_account_ids}")
150 | print(f"📊 Expected accounts found: {results[query]['found_expected_accounts']}/{results[query]['total_expected']}")
151 | print(f"🔍 Raw response: {json.dumps(parsed_content, indent=2)}")
152 |
153 | except json.JSONDecodeError:
154 | results[query] = {
155 | "success": False,
156 | "error": "Search tool did not return valid JSON",
157 | "raw_content": content
158 | }
159 | print(f"❌ Invalid JSON response: {content}")
160 |
161 | return results
162 |
163 | def test_fetch_account(self, account_id: str) -> Dict[str, Any]:
164 | """Test fetching a specific account by ID"""
165 |
166 | print(f"\n🔍 Testing fetch for account: {account_id}")
167 |
168 | result = self._make_request("tools/call", {
169 | "name": "fetch",
170 | "arguments": {"id": account_id}
171 | })
172 |
173 | if not result["success"]:
174 | return {
175 | "success": False,
176 | "error": result.get("text", "Unknown error")
177 | }
178 |
179 | # Parse the tool response
180 | response_data = result["json"]["result"]
181 | content = response_data.get("content", [{}])[0].get("text", "")
182 |
183 | try:
184 | parsed_content = json.loads(content)
185 |
186 | # Validate required fields for OpenAI MCP compliance
187 | required_fields = ["id", "title", "text"]
188 | has_required_fields = all(field in parsed_content for field in required_fields)
189 |
190 | result_data = {
191 | "success": True,
192 | "record": parsed_content,
193 | "has_required_fields": has_required_fields,
194 | "missing_fields": [field for field in required_fields if field not in parsed_content]
195 | }
196 |
197 | if has_required_fields:
198 | print(f"✅ Successfully fetched account with all required fields")
199 | print(f" Title: {parsed_content.get('title', 'N/A')}")
200 | print(f" ID: {parsed_content.get('id', 'N/A')}")
201 | else:
202 | print(f"⚠️ Account fetched but missing required fields: {result_data['missing_fields']}")
203 |
204 | return result_data
205 |
206 | except json.JSONDecodeError:
207 | return {
208 | "success": False,
209 | "error": "Fetch tool did not return valid JSON",
210 | "raw_content": content
211 | }
212 |
213 | def run_account_search_tests(self) -> bool:
214 | """Run comprehensive account search tests"""
215 |
216 | print("🚀 Meta Ads Account Search Test Suite")
217 | print("="*50)
218 |
219 | # Check server availability
220 | try:
221 | response = requests.get(f"{self.base_url}/", timeout=5)
222 | server_running = response.status_code in [200, 404]
223 | except:
224 | server_running = False
225 |
226 | if not server_running:
227 | print("❌ Server is not running at", self.base_url)
228 | print(" Please start the server with:")
229 | print(" python3 -m meta_ads_mcp --transport streamable-http --port 8080")
230 | return False
231 |
232 | print("✅ Server is running")
233 | print("🔐 Using implicit authentication from server")
234 |
235 | # Test 0: First try get_ad_accounts to see if we can get raw data
236 | print("\n" + "="*50)
237 | print("📋 PHASE 0: Testing Direct Account Access")
238 | print("="*50)
239 |
240 | account_result = self._make_request("tools/call", {
241 | "name": "get_ad_accounts",
242 | "arguments": {
243 | "user_id": "me",
244 | "parameters": json.dumps({"limit": 5})
245 | }
246 | })
247 |
248 | if account_result["success"]:
249 | response_data = account_result["json"]["result"]
250 | content = response_data.get("content", [{}])[0].get("text", "")
251 | try:
252 | account_data = json.loads(content)
253 | print(f"✅ get_ad_accounts returned: {json.dumps(account_data, indent=2)}")
254 | except:
255 | print(f"⚠️ get_ad_accounts raw response: {content}")
256 | else:
257 | print(f"❌ get_ad_accounts failed: {account_result.get('text', 'Unknown error')}")
258 |
259 | # Test 1: Search for accounts
260 | print("\n" + "="*50)
261 | print("📋 PHASE 1: Testing Account Search")
262 | print("="*50)
263 |
264 | search_results = self.test_search_accounts()
265 |
266 | # Find the best search result that returned accounts
267 | best_search = None
268 | for query, result in search_results.items():
269 | if result.get("success") and result.get("account_ids"):
270 | best_search = result
271 | break
272 |
273 | if not best_search:
274 | print("\n❌ No search queries returned account IDs")
275 | print("📊 Search Results Summary:")
276 | for query, result in search_results.items():
277 | if result.get("success"):
278 | print(f" '{query}': {len(result.get('ids', []))} total IDs, {len(result.get('account_ids', []))} account IDs")
279 | else:
280 | print(f" '{query}': FAILED - {result.get('error', 'Unknown error')}")
281 | return False
282 |
283 | print(f"\n✅ Found accounts in search results")
284 | account_ids = best_search["account_ids"]
285 | print(f"📋 Account IDs found: {account_ids}")
286 |
287 | # Test 2: Fetch account details
288 | print("\n" + "="*50)
289 | print("📋 PHASE 2: Testing Account Fetch")
290 | print("="*50)
291 |
292 | fetch_success = True
293 | for account_id in account_ids[:2]: # Test first 2 accounts
294 | fetch_result = self.test_fetch_account(account_id)
295 | if not fetch_result["success"]:
296 | print(f"❌ Failed to fetch {account_id}: {fetch_result.get('error', 'Unknown error')}")
297 | fetch_success = False
298 | elif not fetch_result["has_required_fields"]:
299 | print(f"⚠️ {account_id} missing required fields: {fetch_result['missing_fields']}")
300 | fetch_success = False
301 |
302 | # Final assessment
303 | print("\n" + "="*50)
304 | print("📊 FINAL RESULTS")
305 | print("="*50)
306 |
307 | if fetch_success and account_ids:
308 | print("✅ Account search and fetch workflow: SUCCESS")
309 | print(f" • Found {len(account_ids)} accounts")
310 | print(f" • All fetched accounts have required fields")
311 | print(f" • OpenAI MCP compliance: PASSED")
312 | return True
313 | else:
314 | print("❌ Account search and fetch workflow: FAILED")
315 | if not account_ids:
316 | print(" • Issue: No account IDs returned by search")
317 | if not fetch_success:
318 | print(" • Issue: Some accounts failed to fetch or missing required fields")
319 | return False
320 |
321 |
322 | def main():
323 | """Main test execution"""
324 | tester = AccountSearchTester()
325 | success = tester.run_account_search_tests()
326 |
327 | if success:
328 | print("\n🎉 All account search tests passed!")
329 | else:
330 | print("\n⚠️ Some account search tests failed - see details above")
331 |
332 | sys.exit(0 if success else 1)
333 |
334 |
335 | if __name__ == "__main__":
336 | main()
```
--------------------------------------------------------------------------------
/tests/test_page_discovery_integration.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Integration tests for page discovery functionality.
3 | Tests the complete workflow from page discovery to ad creative creation.
4 | """
5 |
6 | import pytest
7 | import json
8 | from unittest.mock import AsyncMock, patch
9 | from meta_ads_mcp.core.ads import create_ad_creative, search_pages_by_name, get_account_pages
10 |
11 |
12 | class TestPageDiscoveryIntegration:
13 | """Integration tests for page discovery functionality."""
14 |
15 | @pytest.mark.asyncio
16 | async def test_end_to_end_page_discovery_in_create_ad_creative(self):
17 | """Test that create_ad_creative automatically discovers pages when no page_id is provided."""
18 | # Mock the page discovery to return a successful result
19 | mock_discovery_result = {
20 | "success": True,
21 | "page_id": "123456789",
22 | "page_name": "Test Page",
23 | "source": "tracking_specs"
24 | }
25 |
26 | # Mock the API request for creating the creative (will fail due to invalid image, but that's expected)
27 | mock_creative_response = {
28 | "error": {
29 | "message": "Invalid parameter",
30 | "type": "OAuthException",
31 | "code": 100,
32 | "error_subcode": 2446386,
33 | "error_user_title": "Image Not Found",
34 | "error_user_msg": "The image you selected is not available."
35 | }
36 | }
37 |
38 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover, \
39 | patch('meta_ads_mcp.core.ads.make_api_request') as mock_api, \
40 | patch('meta_ads_mcp.core.auth.get_current_access_token') as mock_get_token:
41 |
42 | # Provide a valid access token to bypass authentication
43 | mock_get_token.return_value = "test_token_123"
44 |
45 | mock_discover.return_value = mock_discovery_result
46 | mock_api.return_value = mock_creative_response
47 |
48 | # Call create_ad_creative without providing page_id
49 | result = await create_ad_creative(
50 | account_id="act_123456789",
51 | name="Test Creative",
52 | image_hash="test_hash_123",
53 | message="Test message",
54 | access_token="test_token_123" # Provide explicit token
55 | )
56 |
57 | result_data = json.loads(result)
58 |
59 | # Handle MCP wrapper - check if result is wrapped in 'data' field
60 | if "data" in result_data:
61 | actual_result = json.loads(result_data["data"])
62 | else:
63 | actual_result = result_data
64 |
65 | # Verify that the function attempted to create a creative (even though it failed due to invalid image)
66 | assert "error" in actual_result
67 | assert "Image Not Found" in actual_result["error"]["error_user_title"]
68 |
69 | # Verify that page discovery was called
70 | mock_discover.assert_called_once_with("act_123456789", "test_token_123")
71 |
72 | @pytest.mark.asyncio
73 | async def test_search_pages_by_name_integration(self):
74 | """Test the complete search_pages_by_name function with real-like data."""
75 | # Mock the page discovery to return a successful result
76 | mock_discovery_result = {
77 | "success": True,
78 | "page_id": "123456789",
79 | "page_name": "Injury Payouts",
80 | "source": "tracking_specs"
81 | }
82 |
83 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover, \
84 | patch('meta_ads_mcp.core.auth.get_current_access_token') as mock_get_token:
85 |
86 | # Provide a valid access token to bypass authentication
87 | mock_get_token.return_value = "test_token_123"
88 |
89 | mock_discover.return_value = mock_discovery_result
90 |
91 | # Test searching for pages
92 | result = await search_pages_by_name(
93 | account_id="act_123456789",
94 | search_term="Injury",
95 | access_token="test_token_123" # Provide explicit token
96 | )
97 |
98 | result_data = json.loads(result)
99 |
100 | # Handle MCP wrapper - check if result is wrapped in 'data' field
101 | if "data" in result_data and isinstance(result_data["data"], str):
102 | actual_result = json.loads(result_data["data"])
103 | else:
104 | actual_result = result_data
105 |
106 | # Verify the search results
107 | assert len(actual_result["data"]) == 1
108 | assert actual_result["data"][0]["id"] == "123456789"
109 | assert actual_result["data"][0]["name"] == "Injury Payouts"
110 | assert actual_result["search_term"] == "Injury"
111 | assert actual_result["total_found"] == 1
112 | assert actual_result["total_available"] == 1
113 |
114 | @pytest.mark.asyncio
115 | async def test_create_ad_creative_with_manual_page_id(self):
116 | """Test that create_ad_creative works with manually provided page_id."""
117 | # Mock the API request for creating the creative
118 | mock_creative_response = {
119 | "error": {
120 | "message": "Invalid parameter",
121 | "type": "OAuthException",
122 | "code": 100,
123 | "error_subcode": 2446386,
124 | "error_user_title": "Image Not Found",
125 | "error_user_msg": "The image you selected is not available."
126 | }
127 | }
128 |
129 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api, \
130 | patch('meta_ads_mcp.core.auth.get_current_access_token') as mock_get_token:
131 |
132 | # Provide a valid access token to bypass authentication
133 | mock_get_token.return_value = "test_token_123"
134 |
135 | mock_api.return_value = mock_creative_response
136 |
137 | # Call create_ad_creative with a manual page_id
138 | result = await create_ad_creative(
139 | account_id="act_123456789",
140 | name="Test Creative",
141 | image_hash="test_hash_123",
142 | page_id="123456789", # Manual page ID
143 | message="Test message",
144 | access_token="test_token_123" # Provide explicit token
145 | )
146 |
147 | result_data = json.loads(result)
148 |
149 | # Handle MCP wrapper - check if result is wrapped in 'data' field
150 | if "data" in result_data:
151 | actual_result = json.loads(result_data["data"])
152 | else:
153 | actual_result = result_data
154 |
155 | # Verify that the function attempted to create a creative
156 | assert "error" in actual_result
157 | assert "Image Not Found" in actual_result["error"]["error_user_title"]
158 |
159 | # Verify that make_api_request was called for creating the creative
160 | mock_api.assert_called_once()
161 |
162 | @pytest.mark.asyncio
163 | async def test_create_ad_creative_no_pages_found(self):
164 | """Test create_ad_creative when no pages are found."""
165 | # Mock the page discovery to return no pages
166 | mock_discovery_result = {
167 | "success": False,
168 | "message": "No suitable pages found for this account"
169 | }
170 |
171 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover, \
172 | patch('meta_ads_mcp.core.auth.get_current_access_token') as mock_get_token:
173 |
174 | # Provide a valid access token to bypass authentication
175 | mock_get_token.return_value = "test_token_123"
176 |
177 | mock_discover.return_value = mock_discovery_result
178 |
179 | # Call create_ad_creative without providing page_id
180 | result = await create_ad_creative(
181 | account_id="act_123456789",
182 | name="Test Creative",
183 | image_hash="test_hash_123",
184 | message="Test message",
185 | access_token="test_token_123" # Provide explicit token
186 | )
187 |
188 | result_data = json.loads(result)
189 |
190 | # Handle MCP wrapper - check if result is wrapped in 'data' field
191 | if "data" in result_data:
192 | actual_result = json.loads(result_data["data"])
193 | else:
194 | actual_result = result_data
195 |
196 | # Verify that the function returned an error about no pages found
197 | assert "error" in actual_result
198 | assert "No page ID provided and no suitable pages found" in actual_result["error"]
199 | assert "suggestions" in actual_result
200 |
201 | @pytest.mark.asyncio
202 | async def test_search_pages_by_name_no_search_term(self):
203 | """Test search_pages_by_name without a search term (should return all pages)."""
204 | # Mock the page discovery to return a successful result
205 | mock_discovery_result = {
206 | "success": True,
207 | "page_id": "123456789",
208 | "page_name": "Test Page",
209 | "source": "tracking_specs"
210 | }
211 |
212 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover, \
213 | patch('meta_ads_mcp.core.auth.get_current_access_token') as mock_get_token:
214 |
215 | # Provide a valid access token to bypass authentication
216 | mock_get_token.return_value = "test_token_123"
217 |
218 | mock_discover.return_value = mock_discovery_result
219 |
220 | # Test searching without a search term
221 | result = await search_pages_by_name(
222 | account_id="act_123456789",
223 | access_token="test_token_123" # Provide explicit token
224 | )
225 |
226 | result_data = json.loads(result)
227 |
228 | # Handle MCP wrapper - check if result is wrapped in 'data' field
229 | if "data" in result_data and isinstance(result_data["data"], str):
230 | actual_result = json.loads(result_data["data"])
231 | else:
232 | actual_result = result_data
233 |
234 | # Verify the results
235 | assert len(actual_result["data"]) == 1
236 | assert actual_result["data"][0]["id"] == "123456789"
237 | assert actual_result["data"][0]["name"] == "Test Page"
238 | assert actual_result["total_available"] == 1
239 | assert "note" in actual_result
240 |
241 | @pytest.mark.asyncio
242 | async def test_search_pages_by_name_no_matches(self):
243 | """Test search_pages_by_name when no pages match the search term."""
244 | # Mock the page discovery to return a successful result
245 | mock_discovery_result = {
246 | "success": True,
247 | "page_id": "123456789",
248 | "page_name": "Test Page",
249 | "source": "tracking_specs"
250 | }
251 |
252 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover, \
253 | patch('meta_ads_mcp.core.auth.get_current_access_token') as mock_get_token:
254 |
255 | # Provide a valid access token to bypass authentication
256 | mock_get_token.return_value = "test_token_123"
257 |
258 | mock_discover.return_value = mock_discovery_result
259 |
260 | # Test searching for a term that doesn't match
261 | result = await search_pages_by_name(
262 | account_id="act_123456789",
263 | search_term="Nonexistent",
264 | access_token="test_token_123" # Provide explicit token
265 | )
266 |
267 | result_data = json.loads(result)
268 |
269 | # Handle MCP wrapper - check if result is wrapped in 'data' field
270 | if "data" in result_data and isinstance(result_data["data"], str):
271 | actual_result = json.loads(result_data["data"])
272 | else:
273 | actual_result = result_data
274 |
275 | # Verify that no pages were found
276 | assert len(actual_result["data"]) == 0
277 | assert actual_result["search_term"] == "Nonexistent"
278 | assert actual_result["total_found"] == 0
279 | assert actual_result["total_available"] == 1
280 |
281 |
282 | if __name__ == "__main__":
283 | pytest.main([__file__])
```
--------------------------------------------------------------------------------
/tests/e2e_account_info_search_issue.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | E2E Test for Account Info Search Issue
4 |
5 | This test reproduces the issue reported by a user where get_account_info
6 | cannot find account ID 414174661097171, while get_ad_accounts can see it.
7 |
8 | Usage:
9 | 1. Start the server: uv run python -m meta_ads_mcp --transport streamable-http --port 8080
10 | 2. Run test: uv run python tests/e2e_account_info_search_issue.py
11 |
12 | Or with pytest (manual only):
13 | uv run python -m pytest tests/e2e_account_info_search_issue.py -v -m e2e
14 | """
15 |
16 | import pytest
17 | import requests
18 | import json
19 | from typing import Dict, Any
20 |
21 | @pytest.mark.e2e
22 | @pytest.mark.skip(reason="E2E test - run manually only")
23 | class TestAccountInfoSearchIssue:
24 | """E2E test for account info search issue"""
25 |
26 | def __init__(self, base_url: str = "http://localhost:8080"):
27 | self.base_url = base_url.rstrip('/')
28 | self.endpoint = f"{self.base_url}/mcp/"
29 | self.request_id = 1
30 | self.target_account_id = "414174661097171"
31 |
32 | def test_get_ad_accounts_can_see_target_account(self):
33 | """Verify get_ad_accounts can see account 414174661097171"""
34 | print(f"\n🔍 Testing if get_ad_accounts can see account {self.target_account_id}")
35 |
36 | params = {
37 | "name": "get_ad_accounts",
38 | "arguments": {}
39 | }
40 |
41 | result = self._make_request("tools/call", params)
42 |
43 | if not result["success"]:
44 | return {
45 | "success": False,
46 | "error": f"Request failed: {result.get('error', 'Unknown error')}",
47 | "status_code": result["status_code"]
48 | }
49 |
50 | try:
51 | response_data = result["json"]["result"]
52 | content = response_data.get("content", [{}])[0].get("text", "")
53 | parsed_content = json.loads(content)
54 |
55 | # Check for errors first
56 | error_info = self._check_for_errors(parsed_content)
57 | if error_info["has_error"]:
58 | return {
59 | "success": False,
60 | "error": f"get_ad_accounts returned error: {error_info['error_message']}",
61 | "error_format": error_info["format"]
62 | }
63 |
64 | # Extract accounts data
65 | accounts = []
66 | if "data" in parsed_content:
67 | data = parsed_content["data"]
68 |
69 | # Handle case where data is already parsed (list/dict)
70 | if isinstance(data, list):
71 | accounts = data
72 | elif isinstance(data, dict) and "data" in data:
73 | accounts = data["data"]
74 |
75 | # Handle case where data is a JSON string that needs parsing
76 | elif isinstance(data, str):
77 | try:
78 | parsed_data = json.loads(data)
79 | if isinstance(parsed_data, list):
80 | accounts = parsed_data
81 | elif isinstance(parsed_data, dict) and "data" in parsed_data:
82 | accounts = parsed_data["data"]
83 | except json.JSONDecodeError:
84 | pass
85 | elif isinstance(parsed_content, list):
86 | accounts = parsed_content
87 |
88 | # Search for target account
89 | found_account = None
90 | for account in accounts:
91 | if isinstance(account, dict):
92 | account_id = account.get("id", "").replace("act_", "")
93 | if account_id == self.target_account_id:
94 | found_account = account
95 | break
96 |
97 | result_data = {
98 | "success": True,
99 | "total_accounts": len(accounts),
100 | "target_account_found": found_account is not None,
101 | "found_account_details": found_account if found_account else None,
102 | "all_account_ids": [
103 | acc.get("id", "").replace("act_", "")
104 | for acc in accounts
105 | if isinstance(acc, dict) and acc.get("id")
106 | ]
107 | }
108 |
109 | print(f"✅ get_ad_accounts results:")
110 | print(f" Total accounts found: {result_data['total_accounts']}")
111 | print(f" Target account {self.target_account_id} found: {result_data['target_account_found']}")
112 | if found_account:
113 | print(f" Account details: {found_account}")
114 |
115 | return result_data
116 |
117 | except json.JSONDecodeError as e:
118 | return {
119 | "success": False,
120 | "error": f"Could not parse get_ad_accounts response: {str(e)}",
121 | "raw_content": content
122 | }
123 |
124 | def test_get_account_info_cannot_find_target_account(self):
125 | """Verify get_account_info cannot find account 414174661097171"""
126 | print(f"\n🔍 Testing if get_account_info can find account {self.target_account_id}")
127 |
128 | params = {
129 | "name": "get_account_info",
130 | "arguments": {
131 | "account_id": self.target_account_id
132 | }
133 | }
134 |
135 | result = self._make_request("tools/call", params)
136 |
137 | if not result["success"]:
138 | return {
139 | "success": False,
140 | "error": f"Request failed: {result.get('error', 'Unknown error')}",
141 | "status_code": result["status_code"]
142 | }
143 |
144 | try:
145 | response_data = result["json"]["result"]
146 | content = response_data.get("content", [{}])[0].get("text", "")
147 | parsed_content = json.loads(content)
148 |
149 | # Check for errors
150 | error_info = self._check_for_errors(parsed_content)
151 |
152 | result_data = {
153 | "success": True,
154 | "has_error": error_info["has_error"],
155 | "error_message": error_info.get("error_message", ""),
156 | "error_format": error_info.get("format", ""),
157 | "raw_response": parsed_content
158 | }
159 |
160 | print(f"✅ get_account_info results:")
161 | print(f" Has error: {result_data['has_error']}")
162 | if result_data['has_error']:
163 | print(f" Error message: {result_data['error_message']}")
164 | print(f" Error format: {result_data['error_format']}")
165 | else:
166 | print(f" Unexpected success: {parsed_content}")
167 |
168 | return result_data
169 |
170 | except json.JSONDecodeError as e:
171 | return {
172 | "success": False,
173 | "error": f"Could not parse get_account_info response: {str(e)}",
174 | "raw_content": content
175 | }
176 |
177 | def _check_for_errors(self, parsed_content: Dict[str, Any]) -> Dict[str, Any]:
178 | """Properly handle both wrapped and direct error formats"""
179 |
180 | # Check for data wrapper format first
181 | if "data" in parsed_content:
182 | data = parsed_content["data"]
183 |
184 | # Handle case where data is already parsed (dict/list)
185 | if isinstance(data, dict) and 'error' in data:
186 | return {
187 | "has_error": True,
188 | "error_message": data['error'],
189 | "error_details": data.get('details', ''),
190 | "format": "wrapped_dict"
191 | }
192 |
193 | # Handle case where data is a JSON string that needs parsing
194 | if isinstance(data, str):
195 | try:
196 | error_data = json.loads(data)
197 | if 'error' in error_data:
198 | return {
199 | "has_error": True,
200 | "error_message": error_data['error'],
201 | "error_details": error_data.get('details', ''),
202 | "format": "wrapped_json"
203 | }
204 | except json.JSONDecodeError:
205 | # Data field exists but isn't valid JSON
206 | pass
207 |
208 | # Check for direct error format
209 | if 'error' in parsed_content:
210 | return {
211 | "has_error": True,
212 | "error_message": parsed_content['error'],
213 | "error_details": parsed_content.get('details', ''),
214 | "format": "direct"
215 | }
216 |
217 | return {"has_error": False}
218 |
219 | def _make_request(self, method: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
220 | """Make a JSON-RPC request to the MCP server"""
221 |
222 | headers = {
223 | "Content-Type": "application/json",
224 | "Accept": "application/json, text/event-stream",
225 | "User-Agent": "E2E-Test-Client/1.0"
226 | }
227 |
228 | payload = {
229 | "jsonrpc": "2.0",
230 | "method": method,
231 | "id": self.request_id
232 | }
233 |
234 | if params:
235 | payload["params"] = params
236 |
237 | try:
238 | response = requests.post(
239 | self.endpoint,
240 | headers=headers,
241 | json=payload,
242 | timeout=30
243 | )
244 |
245 | self.request_id += 1
246 |
247 | return {
248 | "status_code": response.status_code,
249 | "json": response.json() if response.status_code == 200 else None,
250 | "text": response.text,
251 | "success": response.status_code == 200
252 | }
253 |
254 | except requests.exceptions.RequestException as e:
255 | return {
256 | "status_code": 0,
257 | "json": None,
258 | "text": str(e),
259 | "success": False,
260 | "error": str(e)
261 | }
262 |
263 | def run_validation():
264 | """Run the validation tests"""
265 | print("🚀 Starting Account Info Search Issue Validation")
266 | print(f"Target Account ID: 414174661097171")
267 | print("="*60)
268 |
269 | test_instance = TestAccountInfoSearchIssue()
270 |
271 | # Test 1: Check if get_ad_accounts can see the account
272 | accounts_result = test_instance.test_get_ad_accounts_can_see_target_account()
273 |
274 | # Test 2: Check if get_account_info can find the account
275 | account_info_result = test_instance.test_get_account_info_cannot_find_target_account()
276 |
277 | print("\n" + "="*60)
278 | print("📊 VALIDATION SUMMARY")
279 | print("="*60)
280 |
281 | if accounts_result["success"]:
282 | print(f"✅ get_ad_accounts: Found {accounts_result['total_accounts']} total accounts")
283 | if accounts_result["target_account_found"]:
284 | print(f"✅ get_ad_accounts: Target account 414174661097171 IS visible")
285 | else:
286 | print(f"❌ get_ad_accounts: Target account 414174661097171 is NOT visible")
287 | print(f" Available account IDs: {accounts_result.get('all_account_ids', [])}")
288 | else:
289 | print(f"❌ get_ad_accounts: Failed - {accounts_result['error']}")
290 |
291 | if account_info_result["success"]:
292 | if account_info_result["has_error"]:
293 | print(f"❌ get_account_info: Cannot find account (Error: {account_info_result['error_message']})")
294 | else:
295 | print(f"✅ get_account_info: Successfully found account")
296 | else:
297 | print(f"❌ get_account_info: Test failed - {account_info_result['error']}")
298 |
299 | # Determine if issue is confirmed
300 | issue_confirmed = (
301 | accounts_result.get("success", False) and
302 | accounts_result.get("target_account_found", False) and
303 | account_info_result.get("success", False) and
304 | account_info_result.get("has_error", False)
305 | )
306 |
307 | print("\n" + "="*60)
308 | if issue_confirmed:
309 | print("🐛 ISSUE CONFIRMED:")
310 | print(" - get_ad_accounts CAN see the account")
311 | print(" - get_account_info CANNOT find the account")
312 | print(" - This validates the user's complaint")
313 | else:
314 | print("🤔 ISSUE NOT CONFIRMED:")
315 | print(" - The behavior may be different than reported")
316 | print(" - Check individual test results above")
317 | print("="*60)
318 |
319 | if __name__ == "__main__":
320 | run_validation()
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/http_auth_integration.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | FastMCP HTTP Authentication Integration for Meta Ads MCP
3 |
4 | This module provides direct integration with FastMCP to inject authentication
5 | from HTTP headers into the tool execution context.
6 | """
7 |
8 | import asyncio
9 | import contextvars
10 | from typing import Optional
11 | from .utils import logger
12 | import json
13 |
14 | # Use context variables instead of thread-local storage for better async support
15 | _auth_token: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar('auth_token', default=None)
16 | _pipeboard_token: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar('pipeboard_token', default=None)
17 |
18 | class FastMCPAuthIntegration:
19 | """Direct integration with FastMCP for HTTP authentication"""
20 |
21 | @staticmethod
22 | def set_auth_token(token: str) -> None:
23 | """Set authentication token for the current context
24 |
25 | Args:
26 | token: Access token to use for this request
27 | """
28 | _auth_token.set(token)
29 |
30 | @staticmethod
31 | def get_auth_token() -> Optional[str]:
32 | """Get authentication token for the current context
33 |
34 | Returns:
35 | Access token if set, None otherwise
36 | """
37 | return _auth_token.get(None)
38 |
39 | @staticmethod
40 | def set_pipeboard_token(token: str) -> None:
41 | """Set Pipeboard token for the current context
42 |
43 | Args:
44 | token: Pipeboard API token to use for this request
45 | """
46 | _pipeboard_token.set(token)
47 |
48 | @staticmethod
49 | def get_pipeboard_token() -> Optional[str]:
50 | """Get Pipeboard token for the current context
51 |
52 | Returns:
53 | Pipeboard token if set, None otherwise
54 | """
55 | return _pipeboard_token.get(None)
56 |
57 | @staticmethod
58 | def clear_auth_token() -> None:
59 | """Clear authentication token for the current context"""
60 | _auth_token.set(None)
61 |
62 | @staticmethod
63 | def clear_pipeboard_token() -> None:
64 | """Clear Pipeboard token for the current context"""
65 | _pipeboard_token.set(None)
66 |
67 | @staticmethod
68 | def extract_token_from_headers(headers: dict) -> Optional[str]:
69 | """Extract token from HTTP headers
70 |
71 | Args:
72 | headers: HTTP request headers
73 |
74 | Returns:
75 | Token if found, None otherwise
76 | """
77 | # Check for Bearer token in Authorization header (primary method)
78 | auth_header = headers.get('Authorization') or headers.get('authorization')
79 | if auth_header and auth_header.lower().startswith('bearer '):
80 | token = auth_header[7:].strip()
81 | logger.debug("Found Bearer token in Authorization header")
82 | return token
83 |
84 | # Check for direct Meta access token
85 | meta_token = headers.get('X-META-ACCESS-TOKEN') or headers.get('x-meta-access-token')
86 | if meta_token:
87 | return meta_token
88 |
89 | # Check for Pipeboard token (legacy support, to be removed)
90 | pipeboard_token = headers.get('X-PIPEBOARD-API-TOKEN') or headers.get('x-pipeboard-api-token')
91 | if pipeboard_token:
92 | logger.debug("Found Pipeboard token in legacy headers")
93 | return pipeboard_token
94 |
95 | return None
96 |
97 | @staticmethod
98 | def extract_pipeboard_token_from_headers(headers: dict) -> Optional[str]:
99 | """Extract Pipeboard token from HTTP headers
100 |
101 | Args:
102 | headers: HTTP request headers
103 |
104 | Returns:
105 | Pipeboard token if found, None otherwise
106 | """
107 | # Check for Pipeboard token in X-Pipeboard-Token header (duplication API pattern)
108 | pipeboard_token = headers.get('X-Pipeboard-Token') or headers.get('x-pipeboard-token')
109 | if pipeboard_token:
110 | logger.debug("Found Pipeboard token in X-Pipeboard-Token header")
111 | return pipeboard_token
112 |
113 | # Check for legacy Pipeboard token header
114 | legacy_token = headers.get('X-PIPEBOARD-API-TOKEN') or headers.get('x-pipeboard-api-token')
115 | if legacy_token:
116 | logger.debug("Found Pipeboard token in legacy X-PIPEBOARD-API-TOKEN header")
117 | return legacy_token
118 |
119 | return None
120 |
121 | def patch_fastmcp_server(mcp_server):
122 | """Patch FastMCP server to inject authentication from HTTP headers
123 |
124 | Args:
125 | mcp_server: FastMCP server instance to patch
126 | """
127 | logger.info("Patching FastMCP server for HTTP authentication")
128 |
129 | # Store the original run method
130 | original_run = mcp_server.run
131 |
132 | def patched_run(transport="stdio", **kwargs):
133 | """Enhanced run method that sets up HTTP auth integration"""
134 | logger.debug(f"Starting FastMCP with transport: {transport}")
135 |
136 | if transport == "streamable-http":
137 | logger.debug("Setting up HTTP authentication for streamable-http transport")
138 | setup_http_auth_patching()
139 |
140 | # Call the original run method
141 | return original_run(transport=transport, **kwargs)
142 |
143 | # Replace the run method
144 | mcp_server.run = patched_run
145 | logger.info("FastMCP server patching complete")
146 |
147 | def setup_http_auth_patching():
148 | """Setup HTTP authentication patching for auth system"""
149 | logger.info("Setting up HTTP authentication patching")
150 |
151 | # Import and patch the auth system
152 | from . import auth
153 | from . import api
154 | from . import authentication
155 |
156 | # Store the original function
157 | original_get_current_access_token = auth.get_current_access_token
158 |
159 | async def get_current_access_token_with_http_support() -> Optional[str]:
160 | """Enhanced get_current_access_token that checks HTTP context first"""
161 |
162 | # Check for context-scoped token first
163 | context_token = FastMCPAuthIntegration.get_auth_token()
164 | if context_token:
165 | return context_token
166 |
167 | # Fall back to original implementation
168 | return await original_get_current_access_token()
169 |
170 | # Replace the function in all modules that imported it
171 | auth.get_current_access_token = get_current_access_token_with_http_support
172 | api.get_current_access_token = get_current_access_token_with_http_support
173 | authentication.get_current_access_token = get_current_access_token_with_http_support
174 |
175 | logger.info("Auth system patching complete - patched in auth, api, and authentication modules")
176 |
177 | # Global instance for easy access
178 | fastmcp_auth = FastMCPAuthIntegration()
179 |
180 | # Forward declaration of setup_starlette_middleware
181 | def setup_starlette_middleware(app):
182 | pass
183 |
184 | def setup_fastmcp_http_auth(mcp_server):
185 | """Setup HTTP authentication integration with FastMCP
186 |
187 | Args:
188 | mcp_server: FastMCP server instance to configure
189 | """
190 | logger.info("Setting up FastMCP HTTP authentication integration")
191 |
192 | # 1. Patch FastMCP's run method to ensure our get_current_access_token patch is applied
193 | # This remains crucial for the token to be picked up by tool calls.
194 | patch_fastmcp_server(mcp_server) # This patches mcp_server.run
195 |
196 | # 2. Patch the methods that provide the Starlette app instance
197 | # This ensures our middleware is added to the app Uvicorn will actually serve.
198 |
199 | app_provider_methods = []
200 | if mcp_server.settings.json_response:
201 | if hasattr(mcp_server, "streamable_http_app") and callable(mcp_server.streamable_http_app):
202 | app_provider_methods.append("streamable_http_app")
203 | else:
204 | logger.warning("mcp_server.streamable_http_app not found or not callable, cannot patch for JSON responses.")
205 | else: # SSE
206 | if hasattr(mcp_server, "sse_app") and callable(mcp_server.sse_app):
207 | app_provider_methods.append("sse_app")
208 | else:
209 | logger.warning("mcp_server.sse_app not found or not callable, cannot patch for SSE responses.")
210 |
211 | if not app_provider_methods:
212 | logger.error("No suitable app provider method (streamable_http_app or sse_app) found on mcp_server. Cannot add HTTP Auth middleware.")
213 | # Fallback or error handling might be needed here if this is critical
214 |
215 | for method_name in app_provider_methods:
216 | original_app_provider_method = getattr(mcp_server, method_name)
217 |
218 | def new_patched_app_provider_method(*args, **kwargs):
219 | # Call the original method to get/create the Starlette app
220 | app = original_app_provider_method(*args, **kwargs)
221 | if app:
222 | logger.debug(f"Original {method_name} returned app: {type(app)}. Adding AuthInjectionMiddleware.")
223 | # Now, add our middleware to this specific app instance
224 | setup_starlette_middleware(app)
225 | else:
226 | logger.error(f"Original {method_name} returned None or a non-app object.")
227 | return app
228 |
229 | setattr(mcp_server, method_name, new_patched_app_provider_method)
230 | logger.debug(f"Patched mcp_server.{method_name} to inject AuthInjectionMiddleware.")
231 |
232 | # The old setup_request_middleware call is no longer needed here,
233 | # as middleware addition is now handled by patching the app provider methods.
234 | # try:
235 | # setup_request_middleware(mcp_server)
236 | # except Exception as e:
237 | # logger.warning(f"Could not setup request middleware: {e}")
238 |
239 | logger.info("FastMCP HTTP authentication integration setup attempt complete.")
240 |
241 | # Remove the old setup_request_middleware function as its logic is integrated above
242 | # def setup_request_middleware(mcp_server): ... (delete this function)
243 |
244 | # --- AuthInjectionMiddleware definition ---
245 | from starlette.middleware.base import BaseHTTPMiddleware
246 | from starlette.requests import Request
247 | import json # Ensure json is imported if not already at the top
248 |
249 | class AuthInjectionMiddleware(BaseHTTPMiddleware):
250 | async def dispatch(self, request: Request, call_next):
251 | logger.debug(f"HTTP Auth Middleware: Processing request to {request.url.path}")
252 | logger.debug(f"HTTP Auth Middleware: Request headers: {list(request.headers.keys())}")
253 |
254 | # Extract both types of tokens for dual-header authentication
255 | auth_token = FastMCPAuthIntegration.extract_token_from_headers(dict(request.headers))
256 | pipeboard_token = FastMCPAuthIntegration.extract_pipeboard_token_from_headers(dict(request.headers))
257 |
258 | if auth_token:
259 | logger.debug(f"HTTP Auth Middleware: Extracted auth token: {auth_token[:10]}...")
260 | logger.debug("Injecting auth token into request context")
261 | FastMCPAuthIntegration.set_auth_token(auth_token)
262 |
263 | if pipeboard_token:
264 | logger.debug(f"HTTP Auth Middleware: Extracted Pipeboard token: {pipeboard_token[:10]}...")
265 | logger.debug("Injecting Pipeboard token into request context")
266 | FastMCPAuthIntegration.set_pipeboard_token(pipeboard_token)
267 |
268 | if not auth_token and not pipeboard_token:
269 | logger.warning("HTTP Auth Middleware: No authentication tokens found in headers")
270 |
271 | try:
272 | response = await call_next(request)
273 | return response
274 | finally:
275 | # Clear tokens that were set for this request
276 | if auth_token:
277 | FastMCPAuthIntegration.clear_auth_token()
278 | if pipeboard_token:
279 | FastMCPAuthIntegration.clear_pipeboard_token()
280 |
281 | def setup_starlette_middleware(app):
282 | """Add AuthInjectionMiddleware to the Starlette app if not already present.
283 |
284 | Args:
285 | app: Starlette app instance
286 | """
287 | if not app:
288 | logger.error("Cannot setup Starlette middleware, app is None.")
289 | return
290 |
291 | # Check if our specific middleware class is already in the stack
292 | already_added = False
293 | # Starlette's app.middleware is a list of Middleware objects.
294 | # app.user_middleware contains middleware added by app.add_middleware()
295 | for middleware_item in app.user_middleware:
296 | if middleware_item.cls == AuthInjectionMiddleware:
297 | already_added = True
298 | break
299 |
300 | if not already_added:
301 | try:
302 | app.add_middleware(AuthInjectionMiddleware)
303 | logger.info("AuthInjectionMiddleware added to Starlette app successfully.")
304 | except Exception as e:
305 | logger.error(f"Failed to add AuthInjectionMiddleware to Starlette app: {e}", exc_info=True)
306 | else:
307 | logger.debug("AuthInjectionMiddleware already present in Starlette app's middleware stack.")
```
--------------------------------------------------------------------------------
/tests/test_mobile_app_adset_issue.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | E2E Test for mobile app adset creation issue (Issue #008)
4 |
5 | This test validates that the create_adset tool supports required parameters
6 | for mobile app campaigns:
7 | - promoted_object configuration
8 | - destination_type settings
9 | - Conversion event dataset linking
10 | - Custom event type specification
11 |
12 | Expected Meta API error when parameters are missing:
13 | "Select a dataset and conversion event for your ad set (Code 100)"
14 |
15 | Usage (Manual execution only):
16 | 1. Start the server: uv run python -m meta_ads_mcp --transport streamable-http --port 8080
17 | 2. Run test: uv run python tests/test_mobile_app_adset_issue.py
18 |
19 | Or with pytest (explicit E2E flag required):
20 | uv run python -m pytest tests/test_mobile_app_adset_issue.py -v -m e2e
21 |
22 | Note: This test is marked as E2E and will NOT run automatically in CI.
23 | It must be executed manually to validate mobile app campaign functionality.
24 | """
25 |
26 | import pytest
27 | import requests
28 | import json
29 | import time
30 | import sys
31 | import os
32 | from typing import Dict, Any
33 |
34 | # Add project root to path for imports
35 | sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
36 |
37 | class MobileAppAdsetTester:
38 | """Test suite for mobile app adset creation functionality"""
39 |
40 | def __init__(self, base_url: str = "http://localhost:8080"):
41 | self.base_url = base_url.rstrip('/')
42 | self.endpoint = f"{self.base_url}/mcp/"
43 | self.request_id = 1
44 |
45 | def _make_request(self, method: str, params: Dict[str, Any] = None,
46 | headers: Dict[str, str] = None) -> Dict[str, Any]:
47 | """Make a JSON-RPC request to the MCP server"""
48 |
49 | # Default headers for MCP protocol with streamable HTTP transport
50 | default_headers = {
51 | "Content-Type": "application/json",
52 | "Accept": "application/json, text/event-stream",
53 | "User-Agent": "MobileApp-Test-Client/1.0"
54 | }
55 |
56 | if headers:
57 | default_headers.update(headers)
58 |
59 | payload = {
60 | "jsonrpc": "2.0",
61 | "method": method,
62 | "id": self.request_id
63 | }
64 |
65 | if params:
66 | payload["params"] = params
67 |
68 | try:
69 | response = requests.post(
70 | self.endpoint,
71 | headers=default_headers,
72 | json=payload,
73 | timeout=30 # Increased timeout for API calls
74 | )
75 |
76 | self.request_id += 1
77 |
78 | return {
79 | "status_code": response.status_code,
80 | "headers": dict(response.headers),
81 | "json": response.json() if response.status_code == 200 else None,
82 | "text": response.text,
83 | "success": response.status_code == 200
84 | }
85 |
86 | except requests.exceptions.RequestException as e:
87 | return {
88 | "status_code": 0,
89 | "headers": {},
90 | "json": None,
91 | "text": str(e),
92 | "success": False,
93 | "error": str(e)
94 | }
95 |
96 | def test_create_adset_tool_exists(self) -> Dict[str, Any]:
97 | """Test that create_adset tool exists and check its parameters"""
98 | result = self._make_request("tools/list", {})
99 |
100 | if not result["success"]:
101 | return {"success": False, "error": "Failed to get tools list"}
102 |
103 | tools = result["json"]["result"].get("tools", [])
104 | create_adset_tool = next((tool for tool in tools if tool["name"] == "create_adset"), None)
105 |
106 | if not create_adset_tool:
107 | return {"success": False, "error": "create_adset tool not found"}
108 |
109 | # Check if mobile app specific parameters are supported
110 | input_schema = create_adset_tool.get("inputSchema", {})
111 | properties = input_schema.get("properties", {})
112 |
113 | mobile_app_params = ["promoted_object", "destination_type"]
114 | missing_params = []
115 |
116 | for param in mobile_app_params:
117 | if param not in properties:
118 | missing_params.append(param)
119 |
120 | return {
121 | "success": True,
122 | "tool": create_adset_tool,
123 | "missing_mobile_app_params": missing_params,
124 | "has_mobile_app_support": len(missing_params) == 0
125 | }
126 |
127 | def test_reproduce_mobile_app_error(self) -> Dict[str, Any]:
128 | """Reproduce mobile app adset creation error scenario"""
129 |
130 | # Test parameters for mobile app campaign
131 | test_params = {
132 | "name": "create_adset",
133 | "arguments": {
134 | "account_id": "act_123456789012345", # Generic test account
135 | "campaign_id": "120230566078340163", # This will likely be invalid but that's OK for testing
136 | "name": "test mobile app ad set",
137 | "status": "PAUSED",
138 | "targeting": {
139 | "age_max": 65,
140 | "age_min": 18,
141 | "app_install_state": "not_installed",
142 | "geo_locations": {
143 | "countries": ["DE"],
144 | "location_types": ["home", "recent"]
145 | },
146 | "user_device": ["Android_Smartphone", "Android_Tablet"],
147 | "user_os": ["Android"],
148 | "brand_safety_content_filter_levels": ["FACEBOOK_STANDARD", "AN_STANDARD"],
149 | "targeting_automation": {"advantage_audience": 1}
150 | },
151 | "optimization_goal": "APP_INSTALLS",
152 | "billing_event": "IMPRESSIONS"
153 | }
154 | }
155 |
156 | result = self._make_request("tools/call", test_params)
157 |
158 | if not result["success"]:
159 | return {
160 | "success": False,
161 | "error": f"MCP call failed: {result.get('text', 'Unknown error')}"
162 | }
163 |
164 | # Parse the response
165 | response_data = result["json"]["result"]
166 | content = response_data.get("content", [{}])[0].get("text", "")
167 |
168 | try:
169 | parsed_content = json.loads(content)
170 |
171 | # Check if this is an error response
172 | if "error" in parsed_content:
173 | error_details = parsed_content["error"]
174 | if isinstance(error_details, dict) and "details" in error_details:
175 | meta_error = error_details["details"]
176 |
177 | # Check for the specific error we're looking for
178 | if isinstance(meta_error, dict) and "error" in meta_error:
179 | error_code = meta_error["error"].get("code")
180 | error_message = meta_error["error"].get("error_user_msg", "")
181 |
182 | is_dataset_error = (
183 | error_code == 100 and
184 | "conversion event" in error_message.lower()
185 | )
186 |
187 | return {
188 | "success": True,
189 | "reproduced_error": is_dataset_error,
190 | "error_code": error_code,
191 | "error_message": error_message,
192 | "full_response": parsed_content
193 | }
194 |
195 | return {
196 | "success": True,
197 | "reproduced_error": False,
198 | "unexpected_response": parsed_content
199 | }
200 |
201 | except json.JSONDecodeError as e:
202 | return {
203 | "success": False,
204 | "error": f"Failed to parse response: {e}",
205 | "raw_content": content
206 | }
207 |
208 | # Pytest E2E test class - marked to prevent automatic execution
209 | @pytest.mark.e2e
210 | @pytest.mark.skip(reason="E2E test - requires running MCP server - execute manually only")
211 | class TestMobileAppAdsetIssueE2E:
212 | """E2E test for mobile app adset creation functionality (Issue #008)"""
213 |
214 | def setup_method(self):
215 | """Set up test instance"""
216 | self.tester = MobileAppAdsetTester()
217 |
218 | def test_create_adset_tool_has_mobile_app_params(self):
219 | """Test that create_adset tool exists and has mobile app parameters"""
220 | result = self.tester.test_create_adset_tool_exists()
221 |
222 | assert result["success"], f"Tool test failed: {result.get('error', 'Unknown error')}"
223 |
224 | missing_params = result["missing_mobile_app_params"]
225 | has_mobile_support = result["has_mobile_app_support"]
226 |
227 | # Report results but don't fail if parameters are missing (this is what we're testing)
228 | if missing_params:
229 | pytest.skip(f"Missing mobile app parameters: {missing_params}")
230 | else:
231 | # Parameters are present - mobile app support is available
232 | assert has_mobile_support, "Tool should have mobile app support when parameters are present"
233 |
234 | def test_reproduce_mobile_app_error_scenario(self):
235 | """Test reproducing mobile app adset creation error scenario"""
236 | result = self.tester.test_reproduce_mobile_app_error()
237 |
238 | assert result["success"], f"Error reproduction test failed: {result.get('error', 'Unknown error')}"
239 |
240 | # This test is mainly for validation, not assertion
241 | # The actual error depends on authentication and server state
242 | if result.get("reproduced_error"):
243 | print(f"Reproduced error - Code: {result.get('error_code')}, Message: {result.get('error_message')}")
244 | else:
245 | print("Different response received (may indicate parameters are working or auth issues)")
246 |
247 |
248 | def main():
249 | """Run mobile app adset creation tests (manual execution)"""
250 | print("🚀 Mobile App Adset Creation E2E Test")
251 | print("=" * 50)
252 | print("⚠️ This is an E2E test - requires MCP server running on localhost:8080")
253 | print(" Start server with: uv run python -m meta_ads_mcp --transport streamable-http --port 8080")
254 | print()
255 |
256 | tester = MobileAppAdsetTester()
257 |
258 | # Test 1: Check if create_adset tool exists and has mobile app parameters
259 | print("\n🧪 Test 1: Checking create_adset tool parameters...")
260 | tool_test = tester.test_create_adset_tool_exists()
261 |
262 | if tool_test["success"]:
263 | missing_params = tool_test["missing_mobile_app_params"]
264 | if missing_params:
265 | print(f"❌ Missing mobile app parameters: {missing_params}")
266 | print("⚠️ Mobile app campaigns may not work without these parameters")
267 | else:
268 | print("✅ All mobile app parameters are present")
269 | else:
270 | print(f"❌ Tool test failed: {tool_test['error']}")
271 |
272 | # Test 2: Try to reproduce mobile app error scenario
273 | print("\n🧪 Test 2: Testing mobile app campaign creation...")
274 | error_test = tester.test_reproduce_mobile_app_error()
275 |
276 | if error_test["success"]:
277 | if error_test.get("reproduced_error"):
278 | print("✅ Successfully reproduced the error!")
279 | print(f" Error Code: {error_test['error_code']}")
280 | print(f" Error Message: {error_test['error_message']}")
281 | else:
282 | print("⚠️ Error not reproduced - different response received")
283 | if "unexpected_response" in error_test:
284 | print(f" Response: {json.dumps(error_test['unexpected_response'], indent=2)}")
285 | else:
286 | print(f"❌ Error reproduction test failed: {error_test['error']}")
287 |
288 | # Summary
289 | print("\n🏁 TEST SUMMARY")
290 | print("=" * 30)
291 |
292 | if tool_test["success"]:
293 | missing_params = tool_test["missing_mobile_app_params"]
294 | issue_confirmed = len(missing_params) > 0
295 | fix_validated = len(missing_params) == 0
296 |
297 | if fix_validated:
298 | print("✅ MOBILE APP SUPPORT VALIDATED")
299 | print(" All required mobile app parameters are present")
300 | print(" Mobile app campaigns should work correctly!")
301 | elif issue_confirmed:
302 | print("❌ MOBILE APP SUPPORT INCOMPLETE")
303 | print(f" Missing parameters: {missing_params}")
304 | print(" Mobile app campaigns may fail without these parameters")
305 | else:
306 | print("❓ STATUS UNCLEAR")
307 | print(" Could not determine mobile app parameter status")
308 | else:
309 | print("❌ TEST FAILED")
310 | print(" Could not connect to MCP server or validate tools")
311 |
312 | return 0
313 |
314 | if __name__ == "__main__":
315 | sys.exit(main())
```
--------------------------------------------------------------------------------
/tests/test_http_transport.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | HTTP Transport Integration Tests for Meta Ads MCP
4 |
5 | This test suite validates the complete HTTP transport functionality including:
6 | - MCP protocol compliance (initialize, tools/list, tools/call)
7 | - Authentication header processing
8 | - JSON-RPC request/response format
9 | - Error handling and validation
10 |
11 | Usage:
12 | 1. Start the server: python -m meta_ads_mcp --transport streamable-http --port 8080
13 | 2. Run tests: python -m pytest tests/test_http_transport.py -v
14 |
15 | Or run directly:
16 | python tests/test_http_transport.py
17 | """
18 |
19 | import requests
20 | import json
21 | import time
22 | import sys
23 | import os
24 | from typing import Dict, Any, Optional
25 |
26 | # Add project root to path for imports
27 | sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
28 |
29 | class HTTPTransportTester:
30 | """Test suite for Meta Ads MCP HTTP transport"""
31 |
32 | def __init__(self, base_url: str = "http://localhost:8080"):
33 | self.base_url = base_url.rstrip('/')
34 | self.endpoint = f"{self.base_url}/mcp/"
35 | self.request_id = 1
36 |
37 | def _make_request(self, method: str, params: Dict[str, Any] = None,
38 | headers: Dict[str, str] = None) -> Dict[str, Any]:
39 | """Make a JSON-RPC request to the MCP server"""
40 |
41 | # Default headers for MCP protocol
42 | default_headers = {
43 | "Content-Type": "application/json",
44 | "Accept": "application/json, text/event-stream",
45 | "User-Agent": "MCP-Test-Client/1.0"
46 | }
47 |
48 | if headers:
49 | default_headers.update(headers)
50 |
51 | payload = {
52 | "jsonrpc": "2.0",
53 | "method": method,
54 | "id": self.request_id
55 | }
56 |
57 | if params:
58 | payload["params"] = params
59 |
60 | try:
61 | response = requests.post(
62 | self.endpoint,
63 | headers=default_headers,
64 | json=payload,
65 | timeout=10
66 | )
67 |
68 | self.request_id += 1
69 |
70 | return {
71 | "status_code": response.status_code,
72 | "headers": dict(response.headers),
73 | "json": response.json() if response.status_code == 200 else None,
74 | "text": response.text,
75 | "success": response.status_code == 200
76 | }
77 |
78 | except requests.exceptions.RequestException as e:
79 | return {
80 | "status_code": 0,
81 | "headers": {},
82 | "json": None,
83 | "text": str(e),
84 | "success": False,
85 | "error": str(e)
86 | }
87 |
88 | def test_server_availability(self) -> bool:
89 | """Test if the server is running and accessible"""
90 | try:
91 | response = requests.get(f"{self.base_url}/", timeout=5)
92 | # We expect a 404 for the root path, but it means the server is running
93 | return response.status_code in [200, 404]
94 | except:
95 | return False
96 |
97 | def test_mcp_initialize(self, auth_headers: Dict[str, str] = None) -> Dict[str, Any]:
98 | """Test MCP initialize method"""
99 | return self._make_request("initialize", {
100 | "protocolVersion": "2024-11-05",
101 | "capabilities": {
102 | "roots": {"listChanged": True},
103 | "sampling": {}
104 | },
105 | "clientInfo": {
106 | "name": "meta-ads-test-client",
107 | "version": "1.0.0"
108 | }
109 | }, auth_headers)
110 |
111 | def test_tools_list(self, auth_headers: Dict[str, str] = None) -> Dict[str, Any]:
112 | """Test tools/list method"""
113 | return self._make_request("tools/list", {}, auth_headers)
114 |
115 | def test_tool_call(self, tool_name: str, arguments: Dict[str, Any] = None,
116 | auth_headers: Dict[str, str] = None) -> Dict[str, Any]:
117 | """Test tools/call method"""
118 | params = {"name": tool_name}
119 | if arguments:
120 | params["arguments"] = arguments
121 |
122 | return self._make_request("tools/call", params, auth_headers)
123 |
124 | def run_protocol_flow_test(self, auth_headers: Dict[str, str] = None,
125 | scenario_name: str = "Default") -> Dict[str, bool]:
126 | """Run complete MCP protocol flow test"""
127 | results = {}
128 |
129 | print(f"\n🧪 Testing: {scenario_name}")
130 | print("="*50)
131 |
132 | # Test 1: Initialize
133 | print("🔍 Testing MCP Initialize Request")
134 | init_result = self.test_mcp_initialize(auth_headers)
135 | results["initialize"] = init_result["success"]
136 |
137 | if not init_result["success"]:
138 | print(f"❌ Initialize failed: {init_result.get('text', 'Unknown error')}")
139 | return results
140 |
141 | print("✅ Initialize successful")
142 | if init_result["json"] and "result" in init_result["json"]:
143 | server_info = init_result["json"]["result"].get("serverInfo", {})
144 | print(f" Server: {server_info.get('name', 'unknown')} v{server_info.get('version', 'unknown')}")
145 |
146 | # Test 2: Tools List
147 | print("\n🔍 Testing Tools List Request")
148 | tools_result = self.test_tools_list(auth_headers)
149 | results["tools_list"] = tools_result["success"]
150 |
151 | if not tools_result["success"]:
152 | print(f"❌ Tools list failed: {tools_result.get('text', 'Unknown error')}")
153 | return results
154 |
155 | print("✅ Tools list successful")
156 | if tools_result["json"] and "result" in tools_result["json"]:
157 | tools = tools_result["json"]["result"].get("tools", [])
158 | print(f" Found {len(tools)} tools")
159 |
160 | # Test 3: Tool Call
161 | print("\n🔍 Testing Tool Call: get_ad_accounts")
162 | tool_result = self.test_tool_call("get_ad_accounts", {"limit": 3}, auth_headers)
163 | results["tool_call"] = tool_result["success"]
164 |
165 | if not tool_result["success"]:
166 | print(f"❌ Tool call failed: {tool_result.get('text', 'Unknown error')}")
167 | return results
168 |
169 | print("✅ Tool call successful")
170 |
171 | # Check if it's an authentication error (expected with test tokens)
172 | if tool_result["json"] and "result" in tool_result["json"]:
173 | content = tool_result["json"]["result"].get("content", [{}])[0].get("text", "")
174 | if "Authentication Required" in content:
175 | print(" 📋 Result: Authentication required (expected with test tokens)")
176 | else:
177 | print(f" 📋 Result: {content[:100]}...")
178 |
179 | print(f"\n📊 Scenario Results:")
180 | print(f" Initialize: {'✅' if results['initialize'] else '❌'}")
181 | print(f" Tools List: {'✅' if results['tools_list'] else '❌'}")
182 | print(f" Tool Call: {'✅' if results['tool_call'] else '❌'}")
183 |
184 | return results
185 |
186 | def run_comprehensive_test_suite(self) -> bool:
187 | """Run complete test suite with multiple authentication scenarios"""
188 | print("🚀 Meta Ads MCP HTTP Transport Test Suite")
189 | print("="*60)
190 |
191 | # Check server availability first
192 | print("🔍 Checking server status...")
193 | if not self.test_server_availability():
194 | print("❌ Server is not running at", self.base_url)
195 | print(" Please start the server with:")
196 | print(" python -m meta_ads_mcp --transport streamable-http --port 8080 --host localhost")
197 | return False
198 |
199 | print("✅ Server is running")
200 |
201 | all_results = {}
202 |
203 | # Test scenarios
204 | scenarios = [
205 | {
206 | "name": "No Authentication",
207 | "headers": None
208 | },
209 | {
210 | "name": "Bearer Token (Primary Path)",
211 | "headers": {"Authorization": "Bearer test_pipeboard_token_12345"}
212 | },
213 | {
214 | "name": "Custom Meta App ID (Fallback Path)",
215 | "headers": {"X-META-APP-ID": "123456789012345"}
216 | },
217 | {
218 | "name": "Both Auth Methods",
219 | "headers": {
220 | "Authorization": "Bearer test_pipeboard_token_12345",
221 | "X-META-APP-ID": "123456789012345"
222 | }
223 | }
224 | ]
225 |
226 | # Run tests for each scenario
227 | for scenario in scenarios:
228 | results = self.run_protocol_flow_test(
229 | auth_headers=scenario["headers"],
230 | scenario_name=scenario["name"]
231 | )
232 | all_results[scenario["name"]] = results
233 |
234 | # Run specific get_ads filtering tests
235 | print("\n🧪 Testing get_ads filtering functionality")
236 | print("="*50)
237 | ads_filter_results = self.test_get_ads_filtering()
238 | all_results["get_ads_filtering"] = ads_filter_results
239 |
240 | # Summary
241 | print("\n🏁 TEST SUITE COMPLETED")
242 | print("="*30)
243 |
244 | all_passed = True
245 | for scenario_name, results in all_results.items():
246 | if isinstance(results, dict):
247 | scenario_success = all(results.values())
248 | else:
249 | scenario_success = results
250 | status = "✅ SUCCESS" if scenario_success else "❌ FAILED"
251 | print(f"{scenario_name}: {status}")
252 | if not scenario_success:
253 | all_passed = False
254 |
255 | print(f"\n📊 Overall Result: {'✅ ALL TESTS PASSED' if all_passed else '❌ SOME TESTS FAILED'}")
256 |
257 | if all_passed:
258 | print("\n🎉 Meta Ads MCP HTTP transport is fully functional!")
259 | print(" • MCP protocol compliance: Complete")
260 | print(" • Authentication integration: Working")
261 | print(" • All tools accessible via HTTP")
262 | print(" • get_ads filtering: Working correctly")
263 | print(" • Ready for production use")
264 |
265 | return all_passed
266 |
267 | def test_get_ads_filtering(self) -> Dict[str, bool]:
268 | """Test get_ads function with different filtering parameters"""
269 | results = {}
270 |
271 | # Test with basic auth headers for these tests
272 | auth_headers = {"Authorization": "Bearer test_pipeboard_token_12345"}
273 |
274 | # Test 1: get_ads without filters (should use account endpoint)
275 | print("🔍 Testing get_ads without filters")
276 | result1 = self.test_tool_call("get_ads", {
277 | "account_id": "act_123456789",
278 | "limit": 5
279 | }, auth_headers)
280 | results["no_filters"] = result1["success"]
281 | if result1["success"]:
282 | print("✅ get_ads without filters successful")
283 | else:
284 | print(f"❌ get_ads without filters failed: {result1.get('text', 'Unknown error')}")
285 |
286 | # Test 2: get_ads with campaign_id filter (should use campaign endpoint)
287 | print("🔍 Testing get_ads with campaign_id filter")
288 | result2 = self.test_tool_call("get_ads", {
289 | "account_id": "act_123456789",
290 | "campaign_id": "123456789012345",
291 | "limit": 5
292 | }, auth_headers)
293 | results["campaign_filter"] = result2["success"]
294 | if result2["success"]:
295 | print("✅ get_ads with campaign_id filter successful")
296 | else:
297 | print(f"❌ get_ads with campaign_id filter failed: {result2.get('text', 'Unknown error')}")
298 |
299 | # Test 3: get_ads with adset_id filter (should use adset endpoint)
300 | print("🔍 Testing get_ads with adset_id filter")
301 | result3 = self.test_tool_call("get_ads", {
302 | "account_id": "act_123456789",
303 | "adset_id": "120228975637820183",
304 | "limit": 5
305 | }, auth_headers)
306 | results["adset_filter"] = result3["success"]
307 | if result3["success"]:
308 | print("✅ get_ads with adset_id filter successful")
309 | else:
310 | print(f"❌ get_ads with adset_id filter failed: {result3.get('text', 'Unknown error')}")
311 |
312 | # Test 4: get_ads with both campaign_id and adset_id (adset_id should take priority)
313 | print("🔍 Testing get_ads with both campaign_id and adset_id (adset_id priority)")
314 | result4 = self.test_tool_call("get_ads", {
315 | "account_id": "act_123456789",
316 | "campaign_id": "123456789012345",
317 | "adset_id": "120228975637820183",
318 | "limit": 5
319 | }, auth_headers)
320 | results["priority_test"] = result4["success"]
321 | if result4["success"]:
322 | print("✅ get_ads priority test successful")
323 | else:
324 | print(f"❌ get_ads priority test failed: {result4.get('text', 'Unknown error')}")
325 |
326 | return results
327 |
328 |
329 | def main():
330 | """Main test execution"""
331 | tester = HTTPTransportTester()
332 | success = tester.run_comprehensive_test_suite()
333 | sys.exit(0 if success else 1)
334 |
335 |
336 | if __name__ == "__main__":
337 | main()
```
--------------------------------------------------------------------------------
/tests/test_update_ad_creative_id.py:
--------------------------------------------------------------------------------
```python
1 | """Tests for update_ad function with creative_id parameter.
2 |
3 | Tests for the enhanced update_ad function that supports:
4 | - Updating ad creative via creative_id parameter
5 | - Combining creative updates with other parameters (status, bid_amount, tracking_specs)
6 | - Error handling for invalid creative_id values
7 | - Validation of creative permissions and compatibility
8 | """
9 |
10 | import pytest
11 | import json
12 | from unittest.mock import AsyncMock, patch
13 | from meta_ads_mcp.core.ads import update_ad
14 |
15 |
16 | @pytest.mark.asyncio
17 | class TestUpdateAdCreativeId:
18 | """Test cases for update_ad function with creative_id parameter."""
19 |
20 | async def test_update_ad_creative_id_success(self):
21 | """Test successfully updating ad with new creative_id."""
22 |
23 | sample_response = {
24 | "success": True,
25 | "id": "test_ad_123"
26 | }
27 |
28 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
29 | mock_api.return_value = sample_response
30 |
31 | result = await update_ad(
32 | ad_id="test_ad_123",
33 | creative_id="new_creative_456",
34 | access_token="test_token"
35 | )
36 |
37 | result_data = json.loads(result)
38 | assert "success" in result_data
39 | assert result_data["success"] is True
40 |
41 | # Verify API was called with correct parameters
42 | mock_api.assert_called_once()
43 | call_args = mock_api.call_args
44 |
45 | # Check endpoint
46 | assert call_args[0][0] == "test_ad_123"
47 |
48 | # Check parameters
49 | params = call_args[0][2] # Third argument is params
50 | assert "creative" in params
51 | creative_data = json.loads(params["creative"])
52 | assert creative_data["creative_id"] == "new_creative_456"
53 |
54 | async def test_update_ad_creative_id_with_other_params(self):
55 | """Test updating creative_id along with status and bid_amount."""
56 |
57 | sample_response = {
58 | "success": True,
59 | "id": "test_ad_123"
60 | }
61 |
62 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
63 | mock_api.return_value = sample_response
64 |
65 | result = await update_ad(
66 | ad_id="test_ad_123",
67 | creative_id="new_creative_456",
68 | status="ACTIVE",
69 | bid_amount=150,
70 | access_token="test_token"
71 | )
72 |
73 | result_data = json.loads(result)
74 | assert "success" in result_data
75 | assert result_data["success"] is True
76 |
77 | # Verify API was called with all parameters
78 | mock_api.assert_called_once()
79 | call_args = mock_api.call_args
80 | params = call_args[0][2]
81 |
82 | assert "creative" in params
83 | assert "status" in params
84 | assert "bid_amount" in params
85 |
86 | creative_data = json.loads(params["creative"])
87 | assert creative_data["creative_id"] == "new_creative_456"
88 | assert params["status"] == "ACTIVE"
89 | assert params["bid_amount"] == "150"
90 |
91 | async def test_update_ad_creative_id_with_tracking_specs(self):
92 | """Test updating creative_id along with tracking_specs."""
93 |
94 | sample_response = {
95 | "success": True,
96 | "id": "test_ad_123"
97 | }
98 |
99 | tracking_specs = [{"action.type": "offsite_conversion", "fb_pixel": ["123456"]}]
100 |
101 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
102 | mock_api.return_value = sample_response
103 |
104 | result = await update_ad(
105 | ad_id="test_ad_123",
106 | creative_id="new_creative_456",
107 | tracking_specs=tracking_specs,
108 | access_token="test_token"
109 | )
110 |
111 | result_data = json.loads(result)
112 | assert "success" in result_data
113 | assert result_data["success"] is True
114 |
115 | # Verify API was called with all parameters
116 | mock_api.assert_called_once()
117 | call_args = mock_api.call_args
118 | params = call_args[0][2]
119 |
120 | assert "creative" in params
121 | assert "tracking_specs" in params
122 |
123 | creative_data = json.loads(params["creative"])
124 | assert creative_data["creative_id"] == "new_creative_456"
125 |
126 | # tracking_specs should be JSON encoded
127 | tracking_data = json.loads(params["tracking_specs"])
128 | assert tracking_data == tracking_specs
129 |
130 | async def test_update_ad_invalid_creative_id(self):
131 | """Test updating ad with invalid creative_id."""
132 |
133 | # Simulate API error for invalid creative_id
134 | api_error = {
135 | "error": {
136 | "message": "Invalid creative ID",
137 | "type": "OAuthException",
138 | "code": 100
139 | }
140 | }
141 |
142 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
143 | mock_api.return_value = api_error
144 |
145 | result = await update_ad(
146 | ad_id="test_ad_123",
147 | creative_id="invalid_creative_999",
148 | access_token="test_token"
149 | )
150 |
151 | result_data = json.loads(result)
152 | # The error might be wrapped in a 'data' field per the memory
153 | if "data" in result_data:
154 | error_data = json.loads(result_data["data"])
155 | assert "error" in error_data
156 | else:
157 | assert "error" in result_data
158 |
159 | async def test_update_ad_nonexistent_creative_id(self):
160 | """Test updating ad with non-existent creative_id."""
161 |
162 | # Simulate API error for non-existent creative_id
163 | api_error = {
164 | "error": {
165 | "message": "Creative does not exist",
166 | "type": "OAuthException",
167 | "code": 803
168 | }
169 | }
170 |
171 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
172 | mock_api.return_value = api_error
173 |
174 | result = await update_ad(
175 | ad_id="test_ad_123",
176 | creative_id="nonexistent_creative_999",
177 | access_token="test_token"
178 | )
179 |
180 | result_data = json.loads(result)
181 | # The error might be wrapped in a 'data' field per the memory
182 | if "data" in result_data:
183 | error_data = json.loads(result_data["data"])
184 | assert "error" in error_data
185 | else:
186 | assert "error" in result_data
187 |
188 | async def test_update_ad_creative_id_from_different_account(self):
189 | """Test updating ad with creative_id from different account."""
190 |
191 | # Simulate API error for cross-account creative access
192 | api_error = {
193 | "error": {
194 | "message": "Creative does not belong to this account",
195 | "type": "OAuthException",
196 | "code": 200
197 | }
198 | }
199 |
200 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
201 | mock_api.return_value = api_error
202 |
203 | result = await update_ad(
204 | ad_id="test_ad_123",
205 | creative_id="other_account_creative_456",
206 | access_token="test_token"
207 | )
208 |
209 | result_data = json.loads(result)
210 | # The error might be wrapped in a 'data' field per the memory
211 | if "data" in result_data:
212 | error_data = json.loads(result_data["data"])
213 | assert "error" in error_data
214 | else:
215 | assert "error" in result_data
216 |
217 | async def test_update_ad_no_parameters(self):
218 | """Test update_ad with no parameters provided."""
219 |
220 | result = await update_ad(
221 | ad_id="test_ad_123",
222 | access_token="test_token"
223 | )
224 |
225 | result_data = json.loads(result)
226 | # The error might be wrapped in a 'data' field
227 | if "data" in result_data:
228 | error_data = json.loads(result_data["data"])
229 | assert "error" in error_data
230 | assert "No update parameters provided" in error_data["error"]
231 | else:
232 | assert "error" in result_data
233 | assert "No update parameters provided" in result_data["error"]
234 |
235 | async def test_update_ad_missing_ad_id(self):
236 | """Test update_ad with missing ad_id."""
237 |
238 | result = await update_ad(
239 | ad_id="",
240 | creative_id="new_creative_456",
241 | access_token="test_token"
242 | )
243 |
244 | result_data = json.loads(result)
245 | # The error might be wrapped in a 'data' field
246 | if "data" in result_data:
247 | error_data = json.loads(result_data["data"])
248 | assert "error" in error_data
249 | assert "Ad ID is required" in error_data["error"]
250 | else:
251 | assert "error" in result_data
252 | assert "Ad ID is required" in result_data["error"]
253 |
254 | async def test_update_ad_creative_id_only(self):
255 | """Test updating only the creative_id without other parameters."""
256 |
257 | sample_response = {
258 | "success": True,
259 | "id": "test_ad_123"
260 | }
261 |
262 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
263 | mock_api.return_value = sample_response
264 |
265 | result = await update_ad(
266 | ad_id="test_ad_123",
267 | creative_id="new_creative_456",
268 | access_token="test_token"
269 | )
270 |
271 | result_data = json.loads(result)
272 | assert "success" in result_data
273 | assert result_data["success"] is True
274 |
275 | # Verify API was called with only creative parameter
276 | mock_api.assert_called_once()
277 | call_args = mock_api.call_args
278 | params = call_args[0][2]
279 |
280 | # Should only have creative parameter, no status or bid_amount
281 | assert "creative" in params
282 | assert "status" not in params
283 | assert "bid_amount" not in params
284 | assert "tracking_specs" not in params
285 |
286 | creative_data = json.loads(params["creative"])
287 | assert creative_data["creative_id"] == "new_creative_456"
288 |
289 | async def test_update_ad_creative_compatibility_validation(self):
290 | """Test creative compatibility validation with different creative types."""
291 |
292 | # This test simulates an API error for incompatible creative types
293 | api_error = {
294 | "error": {
295 | "message": "Creative format not compatible with ad set placement",
296 | "type": "OAuthException",
297 | "code": 1487
298 | }
299 | }
300 |
301 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
302 | mock_api.return_value = api_error
303 |
304 | result = await update_ad(
305 | ad_id="test_ad_123",
306 | creative_id="incompatible_creative_456",
307 | access_token="test_token"
308 | )
309 |
310 | result_data = json.loads(result)
311 | # The error might be wrapped in a 'data' field per the memory
312 | if "data" in result_data:
313 | error_data = json.loads(result_data["data"])
314 | assert "error" in error_data
315 | else:
316 | assert "error" in result_data
317 |
318 | async def test_update_ad_api_error_handling(self):
319 | """Test general API error handling for update_ad."""
320 |
321 | with patch('meta_ads_mcp.core.ads.make_api_request', new_callable=AsyncMock) as mock_api:
322 | # Simulate API request throwing an exception
323 | mock_api.side_effect = Exception("API connection failed")
324 |
325 | result = await update_ad(
326 | ad_id="test_ad_123",
327 | creative_id="new_creative_456",
328 | access_token="test_token"
329 | )
330 |
331 | result_data = json.loads(result)
332 | # The error might be wrapped in a 'data' field
333 | if "data" in result_data:
334 | error_data = json.loads(result_data["data"])
335 | assert "error" in error_data
336 | assert "API connection failed" in error_data["error"]
337 | else:
338 | assert "error" in result_data
339 | assert "API connection failed" in result_data["error"]
```
--------------------------------------------------------------------------------
/meta_ads_mcp/core/campaigns.py:
--------------------------------------------------------------------------------
```python
1 | """Campaign-related functionality for Meta Ads API."""
2 |
3 | import json
4 | from typing import List, Optional, Dict, Any, Union
5 | from .api import meta_api_tool, make_api_request
6 | from .accounts import get_ad_accounts
7 | from .server import mcp_server
8 |
9 |
10 | @mcp_server.tool()
11 | @meta_api_tool
12 | async def get_campaigns(account_id: str, access_token: Optional[str] = None, limit: int = 10, status_filter: str = "", after: str = "") -> str:
13 | """
14 | Get campaigns for a Meta Ads account with optional filtering.
15 |
16 | Note: By default, the Meta API returns a subset of available fields.
17 | Other fields like 'effective_status', 'special_ad_categories',
18 | 'lifetime_budget', 'spend_cap', 'budget_remaining', 'promoted_object',
19 | 'source_campaign_id', etc., might be available but require specifying them
20 | in the API call (currently not exposed by this tool's parameters).
21 |
22 | Args:
23 | account_id: Meta Ads account ID (format: act_XXXXXXXXX)
24 | access_token: Meta API access token (optional - will use cached token if not provided)
25 | limit: Maximum number of campaigns to return (default: 10)
26 | status_filter: Filter by effective status (e.g., 'ACTIVE', 'PAUSED', 'ARCHIVED').
27 | Maps to the 'effective_status' API parameter, which expects an array
28 | (this function handles the required JSON formatting). Leave empty for all statuses.
29 | after: Pagination cursor to get the next set of results
30 | """
31 | # Require explicit account_id
32 | if not account_id:
33 | return json.dumps({"error": "No account ID specified"}, indent=2)
34 |
35 | endpoint = f"{account_id}/campaigns"
36 | params = {
37 | "fields": "id,name,objective,status,daily_budget,lifetime_budget,buying_type,start_time,stop_time,created_time,updated_time,bid_strategy",
38 | "limit": limit
39 | }
40 |
41 | if status_filter:
42 | # API expects an array, encode it as a JSON string
43 | params["effective_status"] = json.dumps([status_filter])
44 |
45 | if after:
46 | params["after"] = after
47 |
48 | data = await make_api_request(endpoint, access_token, params)
49 |
50 | return json.dumps(data, indent=2)
51 |
52 |
53 | @mcp_server.tool()
54 | @meta_api_tool
55 | async def get_campaign_details(campaign_id: str, access_token: Optional[str] = None) -> str:
56 | """
57 | Get detailed information about a specific campaign.
58 |
59 | Note: This function requests a specific set of fields ('id,name,objective,status,...').
60 | The Meta API offers many other fields for campaigns (e.g., 'effective_status', 'source_campaign_id', etc.)
61 | that could be added to the 'fields' parameter in the code if needed.
62 |
63 | Args:
64 | campaign_id: Meta Ads campaign ID
65 | access_token: Meta API access token (optional - will use cached token if not provided)
66 | """
67 | if not campaign_id:
68 | return json.dumps({"error": "No campaign ID provided"}, indent=2)
69 |
70 | endpoint = f"{campaign_id}"
71 | params = {
72 | "fields": "id,name,objective,status,daily_budget,lifetime_budget,buying_type,start_time,stop_time,created_time,updated_time,bid_strategy,special_ad_categories,special_ad_category_country,budget_remaining,configured_status"
73 | }
74 |
75 | data = await make_api_request(endpoint, access_token, params)
76 |
77 | return json.dumps(data, indent=2)
78 |
79 |
80 | @mcp_server.tool()
81 | @meta_api_tool
82 | async def create_campaign(
83 | account_id: str,
84 | name: str,
85 | objective: str,
86 | access_token: Optional[str] = None,
87 | status: str = "PAUSED",
88 | special_ad_categories: Optional[List[str]] = None,
89 | daily_budget: Optional[int] = None,
90 | lifetime_budget: Optional[int] = None,
91 | buying_type: Optional[str] = None,
92 | bid_strategy: Optional[str] = None,
93 | bid_cap: Optional[int] = None,
94 | spend_cap: Optional[int] = None,
95 | campaign_budget_optimization: Optional[bool] = None,
96 | ab_test_control_setups: Optional[List[Dict[str, Any]]] = None,
97 | use_adset_level_budgets: bool = False
98 | ) -> str:
99 | """
100 | Create a new campaign in a Meta Ads account.
101 |
102 | Args:
103 | account_id: Meta Ads account ID (format: act_XXXXXXXXX)
104 | name: Campaign name
105 | objective: Campaign objective (ODAX, outcome-based). Must be one of:
106 | OUTCOME_AWARENESS, OUTCOME_TRAFFIC, OUTCOME_ENGAGEMENT,
107 | OUTCOME_LEADS, OUTCOME_SALES, OUTCOME_APP_PROMOTION.
108 | Note: Legacy objectives like BRAND_AWARENESS, LINK_CLICKS,
109 | CONVERSIONS, APP_INSTALLS, etc. are not valid for new
110 | campaigns and will cause a 400 error. Use the outcome-based
111 | values above (e.g., BRAND_AWARENESS → OUTCOME_AWARENESS).
112 | access_token: Meta API access token (optional - will use cached token if not provided)
113 | status: Initial campaign status (default: PAUSED)
114 | special_ad_categories: List of special ad categories if applicable
115 | daily_budget: Daily budget in account currency (in cents) as a string (only used if use_adset_level_budgets=False)
116 | lifetime_budget: Lifetime budget in account currency (in cents) as a string (only used if use_adset_level_budgets=False)
117 | buying_type: Buying type (e.g., 'AUCTION')
118 | bid_strategy: Bid strategy. Must be one of: 'LOWEST_COST_WITHOUT_CAP', 'LOWEST_COST_WITH_BID_CAP', 'COST_CAP', 'LOWEST_COST_WITH_MIN_ROAS'.
119 | bid_cap: Bid cap in account currency (in cents) as a string
120 | spend_cap: Spending limit for the campaign in account currency (in cents) as a string
121 | campaign_budget_optimization: Whether to enable campaign budget optimization (only used if use_adset_level_budgets=False)
122 | ab_test_control_setups: Settings for A/B testing (e.g., [{"name":"Creative A", "ad_format":"SINGLE_IMAGE"}])
123 | use_adset_level_budgets: If True, budgets will be set at the ad set level instead of campaign level (default: False)
124 | """
125 | # Check required parameters
126 | if not account_id:
127 | return json.dumps({"error": "No account ID provided"}, indent=2)
128 |
129 | if not name:
130 | return json.dumps({"error": "No campaign name provided"}, indent=2)
131 |
132 | if not objective:
133 | return json.dumps({"error": "No campaign objective provided"}, indent=2)
134 |
135 | # Special_ad_categories is required by the API, set default if not provided
136 | if special_ad_categories is None:
137 | special_ad_categories = []
138 |
139 | # For this example, we'll add a fixed daily budget if none is provided and we're not using ad set level budgets
140 | if not daily_budget and not lifetime_budget and not use_adset_level_budgets:
141 | daily_budget = "1000" # Default to $10 USD
142 |
143 | endpoint = f"{account_id}/campaigns"
144 |
145 | params = {
146 | "name": name,
147 | "objective": objective,
148 | "status": status,
149 | "special_ad_categories": json.dumps(special_ad_categories) # Properly format as JSON string
150 | }
151 |
152 | # Only set campaign-level budgets if we're not using ad set level budgets
153 | if not use_adset_level_budgets:
154 | # Convert budget values to strings if they aren't already
155 | if daily_budget is not None:
156 | params["daily_budget"] = str(daily_budget)
157 |
158 | if lifetime_budget is not None:
159 | params["lifetime_budget"] = str(lifetime_budget)
160 |
161 | if campaign_budget_optimization is not None:
162 | params["campaign_budget_optimization"] = "true" if campaign_budget_optimization else "false"
163 |
164 | # Add new parameters
165 | if buying_type:
166 | params["buying_type"] = buying_type
167 |
168 | if bid_strategy:
169 | params["bid_strategy"] = bid_strategy
170 |
171 | if bid_cap is not None:
172 | params["bid_cap"] = str(bid_cap)
173 |
174 | if spend_cap is not None:
175 | params["spend_cap"] = str(spend_cap)
176 |
177 | if ab_test_control_setups:
178 | params["ab_test_control_setups"] = json.dumps(ab_test_control_setups)
179 |
180 | try:
181 | data = await make_api_request(endpoint, access_token, params, method="POST")
182 |
183 | # Add a note about budget strategy if using ad set level budgets
184 | if use_adset_level_budgets:
185 | data["budget_strategy"] = "ad_set_level"
186 | data["note"] = "Campaign created with ad set level budgets. Set budgets when creating ad sets within this campaign."
187 |
188 | return json.dumps(data, indent=2)
189 | except Exception as e:
190 | error_msg = str(e)
191 | return json.dumps({
192 | "error": "Failed to create campaign",
193 | "details": error_msg,
194 | "params_sent": params
195 | }, indent=2)
196 |
197 |
198 | @mcp_server.tool()
199 | @meta_api_tool
200 | async def update_campaign(
201 | campaign_id: str,
202 | access_token: Optional[str] = None,
203 | name: Optional[str] = None,
204 | status: Optional[str] = None,
205 | special_ad_categories: Optional[List[str]] = None,
206 | daily_budget: Optional[int] = None,
207 | lifetime_budget: Optional[int] = None,
208 | bid_strategy: Optional[str] = None,
209 | bid_cap: Optional[int] = None,
210 | spend_cap: Optional[int] = None,
211 | campaign_budget_optimization: Optional[bool] = None,
212 | objective: Optional[str] = None, # Add objective if it's updatable
213 | use_adset_level_budgets: Optional[bool] = None, # Add other updatable fields as needed based on API docs
214 | ) -> str:
215 | """
216 | Update an existing campaign in a Meta Ads account.
217 |
218 | Args:
219 | campaign_id: Meta Ads campaign ID
220 | access_token: Meta API access token (optional - will use cached token if not provided)
221 | name: New campaign name
222 | status: New campaign status (e.g., 'ACTIVE', 'PAUSED')
223 | special_ad_categories: List of special ad categories if applicable
224 | daily_budget: New daily budget in account currency (in cents) as a string.
225 | Set to empty string "" to remove the daily budget.
226 | lifetime_budget: New lifetime budget in account currency (in cents) as a string.
227 | Set to empty string "" to remove the lifetime budget.
228 | bid_strategy: New bid strategy
229 | bid_cap: New bid cap in account currency (in cents) as a string
230 | spend_cap: New spending limit for the campaign in account currency (in cents) as a string
231 | campaign_budget_optimization: Enable/disable campaign budget optimization
232 | objective: New campaign objective (Note: May not always be updatable)
233 | use_adset_level_budgets: If True, removes campaign-level budgets to switch to ad set level budgets
234 | """
235 | if not campaign_id:
236 | return json.dumps({"error": "No campaign ID provided"}, indent=2)
237 |
238 | endpoint = f"{campaign_id}"
239 |
240 | params = {}
241 |
242 | # Add parameters to the request only if they are provided
243 | if name is not None:
244 | params["name"] = name
245 | if status is not None:
246 | params["status"] = status
247 | if special_ad_categories is not None:
248 | # Note: Updating special_ad_categories might have specific API rules or might not be allowed after creation.
249 | # The API might require an empty list `[]` to clear categories. Check Meta Docs.
250 | params["special_ad_categories"] = json.dumps(special_ad_categories)
251 |
252 | # Handle budget parameters based on use_adset_level_budgets setting
253 | if use_adset_level_budgets is not None:
254 | if use_adset_level_budgets:
255 | # Remove campaign-level budgets when switching to ad set level budgets
256 | params["daily_budget"] = ""
257 | params["lifetime_budget"] = ""
258 | if campaign_budget_optimization is not None:
259 | params["campaign_budget_optimization"] = "false"
260 | else:
261 | # If switching back to campaign-level budgets, use the provided budget values
262 | if daily_budget is not None:
263 | if daily_budget == "":
264 | params["daily_budget"] = ""
265 | else:
266 | params["daily_budget"] = str(daily_budget)
267 | if lifetime_budget is not None:
268 | if lifetime_budget == "":
269 | params["lifetime_budget"] = ""
270 | else:
271 | params["lifetime_budget"] = str(lifetime_budget)
272 | if campaign_budget_optimization is not None:
273 | params["campaign_budget_optimization"] = "true" if campaign_budget_optimization else "false"
274 | else:
275 | # Normal budget updates when not changing budget strategy
276 | if daily_budget is not None:
277 | # To remove budget, set to empty string
278 | if daily_budget == "":
279 | params["daily_budget"] = ""
280 | else:
281 | params["daily_budget"] = str(daily_budget)
282 | if lifetime_budget is not None:
283 | # To remove budget, set to empty string
284 | if lifetime_budget == "":
285 | params["lifetime_budget"] = ""
286 | else:
287 | params["lifetime_budget"] = str(lifetime_budget)
288 | if campaign_budget_optimization is not None:
289 | params["campaign_budget_optimization"] = "true" if campaign_budget_optimization else "false"
290 |
291 | if bid_strategy is not None:
292 | params["bid_strategy"] = bid_strategy
293 | if bid_cap is not None:
294 | params["bid_cap"] = str(bid_cap)
295 | if spend_cap is not None:
296 | params["spend_cap"] = str(spend_cap)
297 | if objective is not None:
298 | params["objective"] = objective # Caution: Objective changes might reset learning or be restricted
299 |
300 | if not params:
301 | return json.dumps({"error": "No update parameters provided"}, indent=2)
302 |
303 | try:
304 | # Use POST method for updates as per Meta API documentation
305 | data = await make_api_request(endpoint, access_token, params, method="POST")
306 |
307 | # Add a note about budget strategy if switching to ad set level budgets
308 | if use_adset_level_budgets is not None and use_adset_level_budgets:
309 | data["budget_strategy"] = "ad_set_level"
310 | data["note"] = "Campaign updated to use ad set level budgets. Set budgets when creating ad sets within this campaign."
311 |
312 | return json.dumps(data, indent=2)
313 | except Exception as e:
314 | error_msg = str(e)
315 | # Include campaign_id in error for better context
316 | return json.dumps({
317 | "error": f"Failed to update campaign {campaign_id}",
318 | "details": error_msg,
319 | "params_sent": params # Be careful about logging sensitive data if any
320 | }, indent=2)
```
--------------------------------------------------------------------------------
/tests/test_page_discovery.py:
--------------------------------------------------------------------------------
```python
1 | """
2 | Test page discovery functionality for Meta Ads MCP.
3 | """
4 |
5 | import pytest
6 | import json
7 | from unittest.mock import AsyncMock, patch
8 | from meta_ads_mcp.core.ads import _discover_pages_for_account, search_pages_by_name, _search_pages_by_name_core
9 |
10 |
11 | class TestPageDiscovery:
12 | """Test page discovery functionality."""
13 |
14 | @pytest.mark.asyncio
15 | async def test_discover_pages_from_tracking_specs(self):
16 | """Test page discovery from tracking specs (most reliable method)."""
17 | mock_ads_data = {
18 | "data": [
19 | {
20 | "id": "123456789",
21 | "tracking_specs": [
22 | {
23 | "page": ["987654321", "111222333"]
24 | }
25 | ]
26 | }
27 | ]
28 | }
29 |
30 | mock_page_data = {
31 | "id": "987654321",
32 | "name": "Test Page",
33 | "username": "testpage",
34 | "category": "Test Category"
35 | }
36 |
37 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api:
38 | # Mock the ads endpoint call
39 | mock_api.side_effect = [
40 | mock_ads_data, # First call for ads
41 | mock_page_data # Second call for page details
42 | ]
43 |
44 | result = await _discover_pages_for_account("act_123456789", "test_token")
45 |
46 | assert result["success"] is True
47 | # Check that we got one of the expected page IDs (set order is not guaranteed)
48 | assert result["page_id"] in ["987654321", "111222333"]
49 | assert result["page_name"] == "Test Page"
50 | assert result["source"] == "tracking_specs"
51 |
52 | @pytest.mark.asyncio
53 | async def test_discover_pages_from_client_pages(self):
54 | """Test page discovery from client_pages endpoint."""
55 | mock_client_pages_data = {
56 | "data": [
57 | {
58 | "id": "555666777",
59 | "name": "Client Page",
60 | "username": "clientpage"
61 | }
62 | ]
63 | }
64 |
65 | # Mock empty ads data, then client_pages data
66 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api:
67 | mock_api.side_effect = [
68 | {"data": []}, # No ads found
69 | mock_client_pages_data # Client pages found
70 | ]
71 |
72 | result = await _discover_pages_for_account("act_123456789", "test_token")
73 |
74 | assert result["success"] is True
75 | assert result["page_id"] == "555666777"
76 | assert result["page_name"] == "Client Page"
77 | assert result["source"] == "client_pages"
78 |
79 | @pytest.mark.asyncio
80 | async def test_discover_pages_from_assigned_pages(self):
81 | """Test page discovery from assigned_pages endpoint."""
82 | mock_assigned_pages_data = {
83 | "data": [
84 | {
85 | "id": "888999000",
86 | "name": "Assigned Page"
87 | }
88 | ]
89 | }
90 |
91 | # Mock empty responses for first two methods, then assigned_pages
92 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api:
93 | mock_api.side_effect = [
94 | {"data": []}, # No ads found
95 | {"data": []}, # No client pages found
96 | mock_assigned_pages_data # Assigned pages found
97 | ]
98 |
99 | result = await _discover_pages_for_account("act_123456789", "test_token")
100 |
101 | assert result["success"] is True
102 | assert result["page_id"] == "888999000"
103 | assert result["page_name"] == "Assigned Page"
104 | assert result["source"] == "assigned_pages"
105 |
106 | @pytest.mark.asyncio
107 | async def test_discover_pages_no_pages_found(self):
108 | """Test page discovery when no pages are found."""
109 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api:
110 | mock_api.side_effect = [
111 | {"data": []}, # No ads found
112 | {"data": []}, # No client pages found
113 | {"data": []} # No assigned pages found
114 | ]
115 |
116 | result = await _discover_pages_for_account("act_123456789", "test_token")
117 |
118 | assert result["success"] is False
119 | assert "No suitable pages found" in result["message"]
120 |
121 | @pytest.mark.asyncio
122 | async def test_discover_pages_with_invalid_page_ids(self):
123 | """Test page discovery with invalid page IDs in tracking_specs."""
124 | mock_ads_data = {
125 | "data": [
126 | {
127 | "id": "123456789",
128 | "tracking_specs": [
129 | {
130 | "page": ["invalid_id", "not_numeric", "123abc"]
131 | }
132 | ]
133 | }
134 | ]
135 | }
136 |
137 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api:
138 | mock_api.side_effect = [
139 | mock_ads_data, # Ads with invalid page IDs
140 | {"data": []}, # No client pages
141 | {"data": []} # No assigned pages
142 | ]
143 |
144 | result = await _discover_pages_for_account("act_123456789", "test_token")
145 |
146 | assert result["success"] is False
147 | assert "No suitable pages found" in result["message"]
148 |
149 | @pytest.mark.asyncio
150 | async def test_discover_pages_api_error_handling(self):
151 | """Test page discovery with API errors."""
152 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api:
153 | mock_api.side_effect = Exception("API Error")
154 |
155 | result = await _discover_pages_for_account("act_123456789", "test_token")
156 |
157 | assert result["success"] is False
158 | assert "Error during page discovery" in result["message"]
159 |
160 | @pytest.mark.asyncio
161 | async def test_search_pages_by_name_logic(self):
162 | """Test the core search logic without authentication interference."""
163 | # Test the filtering logic directly
164 | mock_pages_data = {
165 | "data": [
166 | {"id": "111", "name": "Test Page 1"},
167 | {"id": "222", "name": "Another Test Page"},
168 | {"id": "333", "name": "Different Page"}
169 | ]
170 | }
171 |
172 | # Test filtering with search term
173 | search_term_lower = "test"
174 | filtered_pages = []
175 |
176 | for page in mock_pages_data["data"]:
177 | page_name = page.get("name", "").lower()
178 | if search_term_lower in page_name:
179 | filtered_pages.append(page)
180 |
181 | assert len(filtered_pages) == 2
182 | assert filtered_pages[0]["name"] == "Test Page 1"
183 | assert filtered_pages[1]["name"] == "Another Test Page"
184 |
185 | @pytest.mark.asyncio
186 | async def test_search_pages_by_name_no_matches(self):
187 | """Test search logic with no matching results."""
188 | mock_pages_data = {
189 | "data": [
190 | {"id": "111", "name": "Test Page 1"},
191 | {"id": "222", "name": "Another Test Page"}
192 | ]
193 | }
194 |
195 | # Test filtering with non-matching search term
196 | search_term_lower = "nonexistent"
197 | filtered_pages = []
198 |
199 | for page in mock_pages_data["data"]:
200 | page_name = page.get("name", "").lower()
201 | if search_term_lower in page_name:
202 | filtered_pages.append(page)
203 |
204 | assert len(filtered_pages) == 0
205 |
206 | @pytest.mark.asyncio
207 | async def test_search_pages_by_name_core_success(self):
208 | """Test the core search function with successful page discovery."""
209 | mock_discovery_result = {
210 | "success": True,
211 | "page_id": "123456789",
212 | "page_name": "Test Page",
213 | "source": "tracking_specs"
214 | }
215 |
216 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover:
217 | mock_discover.return_value = mock_discovery_result
218 |
219 | result = await _search_pages_by_name_core("test_token", "act_123456789", "test")
220 | result_data = json.loads(result)
221 |
222 | assert len(result_data["data"]) == 1
223 | assert result_data["data"][0]["id"] == "123456789"
224 | assert result_data["data"][0]["name"] == "Test Page"
225 | assert result_data["search_term"] == "test"
226 | assert result_data["total_found"] == 1
227 | assert result_data["total_available"] == 1
228 |
229 | @pytest.mark.asyncio
230 | async def test_search_pages_by_name_core_no_pages(self):
231 | """Test the core search function when no pages are found."""
232 | mock_discovery_result = {
233 | "success": False,
234 | "message": "No pages found"
235 | }
236 |
237 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover:
238 | mock_discover.return_value = mock_discovery_result
239 |
240 | result = await _search_pages_by_name_core("test_token", "act_123456789", "test")
241 | result_data = json.loads(result)
242 |
243 | assert len(result_data["data"]) == 0
244 | assert "No pages found" in result_data["message"]
245 |
246 | @pytest.mark.asyncio
247 | async def test_search_pages_by_name_core_no_search_term(self):
248 | """Test the core search function without search term."""
249 | mock_discovery_result = {
250 | "success": True,
251 | "page_id": "123456789",
252 | "page_name": "Test Page",
253 | "source": "tracking_specs"
254 | }
255 |
256 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover:
257 | mock_discover.return_value = mock_discovery_result
258 |
259 | result = await _search_pages_by_name_core("test_token", "act_123456789")
260 | result_data = json.loads(result)
261 |
262 | assert len(result_data["data"]) == 1
263 | assert result_data["total_available"] == 1
264 | assert "note" in result_data
265 |
266 | @pytest.mark.asyncio
267 | async def test_search_pages_by_name_core_exception_handling(self):
268 | """Test the core search function with exception handling."""
269 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover:
270 | mock_discover.side_effect = Exception("Test exception")
271 |
272 | result = await _search_pages_by_name_core("test_token", "act_123456789", "test")
273 | result_data = json.loads(result)
274 |
275 | assert "error" in result_data
276 | assert "Failed to search pages by name" in result_data["error"]
277 |
278 | @pytest.mark.asyncio
279 | async def test_discover_pages_with_multiple_page_ids(self):
280 | """Test page discovery with multiple page IDs in tracking_specs."""
281 | mock_ads_data = {
282 | "data": [
283 | {
284 | "id": "123456789",
285 | "tracking_specs": [
286 | {
287 | "page": ["111222333", "444555666", "777888999"]
288 | }
289 | ]
290 | }
291 | ]
292 | }
293 |
294 | mock_page_data = {
295 | "id": "111222333",
296 | "name": "First Page",
297 | "username": "firstpage"
298 | }
299 |
300 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api:
301 | mock_api.side_effect = [
302 | mock_ads_data, # First call for ads
303 | mock_page_data # Second call for page details
304 | ]
305 |
306 | result = await _discover_pages_for_account("act_123456789", "test_token")
307 |
308 | assert result["success"] is True
309 | # Should get the first page ID from the set
310 | assert result["page_id"] in ["111222333", "444555666", "777888999"]
311 | assert result["page_name"] == "First Page"
312 |
313 | @pytest.mark.asyncio
314 | async def test_discover_pages_with_mixed_valid_invalid_ids(self):
315 | """Test page discovery with mixed valid and invalid page IDs."""
316 | mock_ads_data = {
317 | "data": [
318 | {
319 | "id": "123456789",
320 | "tracking_specs": [
321 | {
322 | "page": ["invalid", "123456789", "not_numeric", "987654321"]
323 | }
324 | ]
325 | }
326 | ]
327 | }
328 |
329 | mock_page_data = {
330 | "id": "123456789",
331 | "name": "Valid Page",
332 | "username": "validpage"
333 | }
334 |
335 | with patch('meta_ads_mcp.core.ads.make_api_request') as mock_api:
336 | mock_api.side_effect = [
337 | mock_ads_data, # First call for ads
338 | mock_page_data # Second call for page details
339 | ]
340 |
341 | result = await _discover_pages_for_account("act_123456789", "test_token")
342 |
343 | assert result["success"] is True
344 | # Should get one of the valid numeric IDs
345 | assert result["page_id"] in ["123456789", "987654321"]
346 | assert result["page_name"] == "Valid Page"
347 |
348 | @pytest.mark.asyncio
349 | async def test_search_pages_by_name_case_insensitive(self):
350 | """Test search function with case insensitive matching."""
351 | mock_discovery_result = {
352 | "success": True,
353 | "page_id": "123456789",
354 | "page_name": "Test Page",
355 | "source": "tracking_specs"
356 | }
357 |
358 | with patch('meta_ads_mcp.core.ads._discover_pages_for_account') as mock_discover:
359 | mock_discover.return_value = mock_discovery_result
360 |
361 | # Test with uppercase search term
362 | result = await _search_pages_by_name_core("test_token", "act_123456789", "TEST")
363 | result_data = json.loads(result)
364 |
365 | assert len(result_data["data"]) == 1
366 | assert result_data["total_found"] == 1
367 |
368 | # Test with lowercase search term
369 | result = await _search_pages_by_name_core("test_token", "act_123456789", "test")
370 | result_data = json.loads(result)
371 |
372 | assert len(result_data["data"]) == 1
373 | assert result_data["total_found"] == 1
374 |
375 |
376 | if __name__ == "__main__":
377 | pytest.main([__file__])
```
--------------------------------------------------------------------------------
/tests/test_account_info_access_fix.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Unit tests for get_account_info accessibility fix.
4 |
5 | This module tests the fix for the issue where get_account_info couldn't access
6 | accounts that were visible in get_ad_accounts but not in the limited direct
7 | accessibility list.
8 |
9 | The fix changes the logic to try fetching account info directly first,
10 | rather than pre-checking against a limited accessibility list.
11 | """
12 |
13 | import pytest
14 | import json
15 | from unittest.mock import AsyncMock, patch
16 |
17 | from meta_ads_mcp.core.accounts import get_account_info
18 |
19 |
20 | class TestAccountInfoAccessFix:
21 | """Test cases for the get_account_info accessibility fix"""
22 |
23 | @pytest.mark.asyncio
24 | async def test_account_info_direct_access_success(self):
25 | """Test that get_account_info works when direct API call succeeds"""
26 |
27 | # Mock the direct account info API response
28 | mock_account_response = {
29 | "id": "act_414174661097171",
30 | "name": "Venture Hunting & Outdoors",
31 | "account_id": "414174661097171",
32 | "account_status": 1,
33 | "amount_spent": "5818510",
34 | "balance": "97677",
35 | "currency": "AUD",
36 | "timezone_name": "Australia/Brisbane",
37 | "business_country_code": "AU"
38 | }
39 |
40 | with patch('meta_ads_mcp.core.accounts.make_api_request', new_callable=AsyncMock) as mock_api:
41 | with patch('meta_ads_mcp.core.auth.get_current_access_token', new_callable=AsyncMock) as mock_auth:
42 | mock_auth.return_value = "test_access_token"
43 | mock_api.return_value = mock_account_response
44 |
45 | result = await get_account_info(account_id="414174661097171")
46 |
47 | # Handle both string and dict return formats
48 | if isinstance(result, str):
49 | result_data = json.loads(result)
50 | else:
51 | result_data = result
52 |
53 | # Verify the account info was returned successfully
54 | assert "error" not in result_data
55 | assert result_data["id"] == "act_414174661097171"
56 | assert result_data["name"] == "Venture Hunting & Outdoors"
57 | assert result_data["account_id"] == "414174661097171"
58 | assert result_data["currency"] == "AUD"
59 | assert result_data["timezone_name"] == "Australia/Brisbane"
60 |
61 | # Verify DSA compliance detection was added
62 | assert "dsa_required" in result_data
63 | assert result_data["dsa_required"] is False # AU is not European
64 | assert "dsa_compliance_note" in result_data
65 |
66 | # Verify the API was called with correct parameters
67 | mock_api.assert_called_once_with(
68 | "act_414174661097171",
69 | "test_access_token",
70 | {
71 | "fields": "id,name,account_id,account_status,amount_spent,balance,currency,age,business_city,business_country_code,timezone_name"
72 | }
73 | )
74 |
75 | @pytest.mark.asyncio
76 | async def test_account_info_permission_error_with_helpful_message(self):
77 | """Test that permission errors provide helpful error messages with accessible accounts"""
78 |
79 | # Mock the permission error response from direct API call
80 | mock_permission_error = {
81 | "error": {
82 | "message": "Insufficient privileges to access the object",
83 | "type": "OAuthException",
84 | "code": 200
85 | }
86 | }
87 |
88 | # Mock accessible accounts response for helpful error message
89 | mock_accessible_accounts = {
90 | "data": [
91 | {"id": "act_123456", "name": "Accessible Account 1"},
92 | {"id": "act_789012", "name": "Accessible Account 2"},
93 | {"id": "act_345678", "name": "Accessible Account 3"}
94 | ]
95 | }
96 |
97 | with patch('meta_ads_mcp.core.accounts.make_api_request', new_callable=AsyncMock) as mock_api:
98 | with patch('meta_ads_mcp.core.auth.get_current_access_token', new_callable=AsyncMock) as mock_auth:
99 | mock_auth.return_value = "test_access_token"
100 |
101 | # First call returns permission error, second call returns accessible accounts
102 | mock_api.side_effect = [mock_permission_error, mock_accessible_accounts]
103 |
104 | result = await get_account_info(account_id="414174661097171")
105 |
106 | # Handle both string and dict return formats
107 | if isinstance(result, str):
108 | result_data = json.loads(result)
109 | else:
110 | result_data = result
111 |
112 | # Verify helpful error message
113 | assert "error" in result_data
114 | assert "not accessible to your user account" in result_data["error"]["message"]
115 | assert "accessible_accounts" in result_data["error"]
116 | assert "suggestion" in result_data["error"]
117 | assert result_data["error"]["total_accessible_accounts"] == 3
118 |
119 | # Verify accessible accounts list
120 | accessible_accounts = result_data["error"]["accessible_accounts"]
121 | assert len(accessible_accounts) == 3
122 | assert accessible_accounts[0]["id"] == "act_123456"
123 | assert accessible_accounts[0]["name"] == "Accessible Account 1"
124 |
125 | # Verify API calls were made
126 | assert mock_api.call_count == 2
127 |
128 | # First call: direct account access attempt
129 | mock_api.assert_any_call(
130 | "act_414174661097171",
131 | "test_access_token",
132 | {
133 | "fields": "id,name,account_id,account_status,amount_spent,balance,currency,age,business_city,business_country_code,timezone_name"
134 | }
135 | )
136 |
137 | # Second call: get accessible accounts for error message
138 | mock_api.assert_any_call(
139 | "me/adaccounts",
140 | "test_access_token",
141 | {
142 | "fields": "id,name,account_id,account_status,amount_spent,balance,currency,age,business_city,business_country_code",
143 | "limit": 50
144 | }
145 | )
146 |
147 | @pytest.mark.asyncio
148 | async def test_account_info_non_permission_error_passthrough(self):
149 | """Test that non-permission errors are passed through unchanged"""
150 |
151 | # Mock a non-permission error (e.g., invalid account ID)
152 | mock_error_response = {
153 | "error": {
154 | "message": "Invalid account ID format",
155 | "type": "GraphAPIException",
156 | "code": 100
157 | }
158 | }
159 |
160 | with patch('meta_ads_mcp.core.accounts.make_api_request', new_callable=AsyncMock) as mock_api:
161 | with patch('meta_ads_mcp.core.auth.get_current_access_token', new_callable=AsyncMock) as mock_auth:
162 | mock_auth.return_value = "test_access_token"
163 | mock_api.return_value = mock_error_response
164 |
165 | result = await get_account_info(account_id="invalid_id")
166 |
167 | # Handle both string and dict return formats
168 | if isinstance(result, str):
169 | result_data = json.loads(result)
170 | else:
171 | result_data = result
172 |
173 | # Verify the original error is returned unchanged
174 | assert result_data == mock_error_response
175 |
176 | # Verify only one API call was made (no attempt to get accessible accounts)
177 | mock_api.assert_called_once()
178 |
179 | @pytest.mark.asyncio
180 | async def test_account_info_missing_account_id_error(self):
181 | """Test that missing account_id parameter returns appropriate error"""
182 |
183 | with patch('meta_ads_mcp.core.auth.get_current_access_token', new_callable=AsyncMock) as mock_auth:
184 | mock_auth.return_value = "test_access_token"
185 |
186 | result = await get_account_info(account_id=None)
187 |
188 | # Handle both string and dict return formats
189 | if isinstance(result, str):
190 | result_data = json.loads(result)
191 | else:
192 | result_data = result
193 |
194 | # Verify error message
195 | assert "error" in result_data
196 | assert "Account ID is required" in result_data["error"]["message"]
197 | assert "Please specify an account_id parameter" in result_data["error"]["details"]
198 |
199 | @pytest.mark.asyncio
200 | async def test_account_info_act_prefix_handling(self):
201 | """Test that account_id prefix handling works correctly"""
202 |
203 | mock_account_response = {
204 | "id": "act_123456789",
205 | "name": "Test Account",
206 | "account_id": "123456789"
207 | }
208 |
209 | with patch('meta_ads_mcp.core.accounts.make_api_request', new_callable=AsyncMock) as mock_api:
210 | with patch('meta_ads_mcp.core.auth.get_current_access_token', new_callable=AsyncMock) as mock_auth:
211 | mock_auth.return_value = "test_access_token"
212 | mock_api.return_value = mock_account_response
213 |
214 | # Test with account ID without act_ prefix
215 | result = await get_account_info(account_id="123456789")
216 |
217 | # Verify the API was called with the act_ prefix added
218 | mock_api.assert_called_once_with(
219 | "act_123456789", # Should have act_ prefix added
220 | "test_access_token",
221 | {
222 | "fields": "id,name,account_id,account_status,amount_spent,balance,currency,age,business_city,business_country_code,timezone_name"
223 | }
224 | )
225 |
226 | @pytest.mark.asyncio
227 | async def test_account_info_european_dsa_detection(self):
228 | """Test that DSA requirements are properly detected for European accounts"""
229 |
230 | # Mock account response for German business
231 | mock_account_response = {
232 | "id": "act_999888777",
233 | "name": "German Test Account",
234 | "account_id": "999888777",
235 | "business_country_code": "DE" # Germany - should trigger DSA
236 | }
237 |
238 | with patch('meta_ads_mcp.core.accounts.make_api_request', new_callable=AsyncMock) as mock_api:
239 | with patch('meta_ads_mcp.core.auth.get_current_access_token', new_callable=AsyncMock) as mock_auth:
240 | mock_auth.return_value = "test_access_token"
241 | mock_api.return_value = mock_account_response
242 |
243 | result = await get_account_info(account_id="999888777")
244 |
245 | # Handle both string and dict return formats
246 | if isinstance(result, str):
247 | result_data = json.loads(result)
248 | else:
249 | result_data = result
250 |
251 | # Verify DSA requirements were properly detected
252 | assert "dsa_required" in result_data
253 | assert result_data["dsa_required"] is True # DE is European
254 | assert "dsa_compliance_note" in result_data
255 | assert "European DSA" in result_data["dsa_compliance_note"]
256 |
257 |
258 | class TestAccountInfoAccessRegression:
259 | """Regression tests to ensure the fix doesn't break existing functionality"""
260 |
261 | @pytest.mark.asyncio
262 | async def test_regression_basic_account_info_still_works(self):
263 | """Regression test: ensure basic account info functionality still works"""
264 |
265 | mock_account_response = {
266 | "id": "act_123456789",
267 | "name": "Basic Test Account",
268 | "account_id": "123456789",
269 | "account_status": 1,
270 | "currency": "USD",
271 | "business_country_code": "US"
272 | }
273 |
274 | with patch('meta_ads_mcp.core.accounts.make_api_request', new_callable=AsyncMock) as mock_api:
275 | with patch('meta_ads_mcp.core.auth.get_current_access_token', new_callable=AsyncMock) as mock_auth:
276 | mock_auth.return_value = "test_access_token"
277 | mock_api.return_value = mock_account_response
278 |
279 | result = await get_account_info(account_id="act_123456789")
280 |
281 | # Handle both string and dict return formats
282 | if isinstance(result, str):
283 | result_data = json.loads(result)
284 | else:
285 | result_data = result
286 |
287 | # Verify basic functionality works
288 | assert "error" not in result_data
289 | assert result_data["id"] == "act_123456789"
290 | assert result_data["name"] == "Basic Test Account"
291 |
292 | def test_account_info_fix_comparison(self):
293 | """
294 | Documentation test: explains what the fix changed
295 |
296 | BEFORE: get_account_info checked accessibility first against limited list (50 accounts)
297 | AFTER: get_account_info tries direct API call first, only shows error if API fails
298 |
299 | This allows accounts visible through business manager (like 414174661097171)
300 | to work properly even if they're not in the limited direct accessibility list.
301 | """
302 |
303 | # This is a documentation test - no actual code execution
304 | old_behavior = "Pre-check accessibility against limited 50 account list"
305 | new_behavior = "Try direct API call first, handle permission errors gracefully"
306 |
307 | assert old_behavior != new_behavior
308 |
309 | # The key insight: get_ad_accounts shows 107 accounts through business manager,
310 | # but "me/adaccounts" only shows 50 directly accessible accounts
311 | total_visible_accounts = 107
312 | directly_accessible_accounts = 50
313 |
314 | assert total_visible_accounts > directly_accessible_accounts
315 |
316 | # Account 414174661097171 was in the 107 but not in the 50
317 | # The fix allows get_account_info to work for such accounts
```