# Marketo API > Implements Marketo API integration (REST, SOAP, Lead Database, Activities, Campaigns, using requests with OAuth 2.0 authentication, lead CRUD, bulk import/export, trigger campaigns, and Marketo REST API patterns. - Skill: `paulpas/marketo-api` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/marketo-api` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/marketo-api/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/marketo-api --- # Marketo API Integration Implements production-grade Marketo integration using the Marketo REST API with OAuth 2.0 authentication. When loaded, this skill makes the model implement operations on the Marketo Lead Database (Leads, Companies, Opportunities), Activities tracking, Campaigns triggers, Bulk API for import/export, Custom Objects, and the Marketo SOAP API for legacy integrations. All implementations follow Marketo best practices: use `MARKETO_CLIENT_ID`, `MARKETO_CLIENT_SECRET`, `MARKETO_BASE_URL` from environment, implement access token caching with auto-refresh, handle rate limits with exponential backoff, use Bulk API for > 300 records, and properly paginate list results using the `nextPageToken`. --- ## Constraints ### MUST DO - Implement structured error responses with consistent format: {error_code, message, details, request_id} - Add rate limiting per client/API key with configurable burst and sustained limits using a token bucket algorithm - Validate all incoming requests against a schema before processing — reject malformed input with clear error messages - Include correlation/request IDs in all log entries for end-to-end request tracing across service boundaries ### MUST NOT DO - Do not expose internal implementation details, stack traces, or database queries in error responses - Avoid accepting unbounded request bodies — set maximum payload sizes and timeout limits - Never trust client-supplied authentication tokens without validation (signature verification, expiration check) - Do not log request/response bodies containing PII, API keys, or other sensitive data ## TL;DR Checklist - [ ] Use Marketo REST API with OAuth 2.0 (client_credentials grant) - [ ] Use `MARKETO_CLIENT_ID`, `MARKETO_CLIENT_SECRET`, `MARKETO_BASE_URL` env vars - [ ] Base URL format: `https://.mktorest.com` - [ ] Cache access tokens (expire in 3600 seconds = 1 hour) - [ ] Leads API: Use `/rest/v1/leads.json` for CRUD operations - [ ] Describe API: Use `/rest/v1/leads/describe.json` for field metadata - [ ] Bulk API: Use for importing/exporting > 300 records - [ ] Activities: Use `/rest/v1/activities.json` with activity type IDs - [ ] Campaigns: Use `/rest/v1/campaigns/{id}/trigger.json` to trigger - [ ] Pagination: Use `nextPageToken` from response for subsequent calls - [ ] Rate limits: 100 calls per 20 seconds per user, varies by tier - [ ] Never log or expose client secret or access tokens --- ## When to Use Use this skill when: - Managing Marketo leads (create, read, update, delete, upsert) - Syncing leads between your app and Marketo - Triggering Marketo campaigns from external systems - Importing/exporting large lead datasets - Querying lead activities (page visits, form fills, email opens, clicks) - Working with Marketo custom objects - Managing Marketo programs and tokens - Building marketing automation integrations - Processing Marketo webhooks - Using Marketo's SOAP API for legacy integrations --- ## When NOT to Use - For HubSpot-specific marketing — use `coding-hubspot-api` instead - For Salesforce CRM — use `coding-salesforce-api` instead - For Zendesk support — use `coding-zendesk-api` instead - For simple HTTP-only use cases when a full SDK wrapper is overkill - When you need real-time events (consider Marketo Webhooks instead) - For read-only analytics reporting (use Marketo Analytics API directly) --- ## Core Workflow 1. **Authenticate** — Get access token using OAuth 2.0 client_credentials grant: - Endpoint: `POST /identity/oauth/token?grant_type=client_credentials&client_id=...&client_secret=...` - Token expires in 3600 seconds (1 hour) — cache it **Checkpoint:** Verify token response contains `access_token` and `expires_in`. 2. **Lead Operations** — Manage leads using REST API: - Get by ID: `GET /rest/v1/leads/{id}.json?fields=...` - Get by filter: `GET /rest/v1/leads.json?filterType=email&filterValues=...&fields=...` - Create/Update: `POST /rest/v1/leads.json` with JSON body - Upsert: Use `action=createOrUpdate` in query params **Checkpoint:** All lead operations specify `fields` parameter to return needed data. 3. **Activities** — Query lead activity history: - Get activity types: `GET /rest/v1/activities/types.json` - Get activities: `GET /rest/v1/activities.json?activityTypeIds=...&nextPageToken=...` - Activity type IDs: 1=Visit Webpage, 2=Fill Out Form, 3=Click Email, etc. **Checkpoint:** Activity queries include `nextPageToken` handling for pagination. 4. **Trigger Campaigns** — Trigger Smart Campaigns via API: - Endpoint: `POST /rest/v1/campaigns/{campaignId}/trigger.json` - Request body: `{"input": [{"id": leadId, "leads": [...]}]}` - Optionally pass `tokens` for dynamic content **Checkpoint:** Campaign exists and is "Trigger Campaign" type, not batch. 5. **Bulk Operations** — Use Bulk API for large datasets: - Bulk Export: Create job → Poll status → Download when complete - Bulk Import: Upload file → Create import job → Poll status - Use for operations involving > 300 records **Checkpoint:** Bulk jobs polled with exponential backoff, not tight loops. 6. **Handle Limits & Rate Limiting** — Handle Marketo's rate limits: - Standard: 100 calls per 20 seconds per user - Concurrent: Max 10 concurrent requests - Daily quota: Varies by subscription tier - 429 or 606 response codes = rate limit exceeded **Checkpoint:** Rate limit handling includes jittered exponential backoff. --- ## Implementation Patterns ### Pattern 1: Marketo Client Initialization (BAD vs GOOD) ```python """Marketo REST API client initialization patterns. Key concepts: - Marketo REST API: Primary API for all modern integrations - Munchkin ID: Unique identifier for your Marketo instance - Base URL: https://.mktorest.com - OAuth 2.0: client_credentials grant type - Access token: Valid for 3600 seconds (1 hour), cache it - Rate limits: 100 calls / 20 sec per user (varies by tier) Environment variables: MARKETO_CLIENT_ID: OAuth client ID from Admin → LaunchPoint MARKETO_CLIENT_SECRET: OAuth client secret MARKETO_BASE_URL: Base URL (e.g., https://123-ABC-456.mktorest.com) MARKETO_MUNCHKIN_ID: Optional (derived from base URL) """ from __future__ import annotations import os import json import time import logging from typing import Any, Optional, List, Dict, TypeVar, Callable, Generator from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from functools import wraps from urllib.parse import urljoin logger = logging.getLogger(__name__) # Try to import requests try: import requests from requests.exceptions import RequestException, HTTPError, Timeout REQUESTS_AVAILABLE = True except ImportError: REQUESTS_AVAILABLE = False logger.warning("requests not installed. Run: pip install requests") # =================================================================== # ❌ BAD — hardcoded credentials, no token caching, no error handling # =================================================================== def bad_marketo_init() -> Any: """❌ BAD: Don't do any of these things.""" if not REQUESTS_AVAILABLE: raise ImportError("requests library required") # ❌ Hardcoded credentials! Never commit these! client_id = "12345678-1234-1234-1234-1234567890ab" client_secret = "abc123def456ghi789" base_url = "https://123-ABC-456.mktorest.com" # ❌ Getting a new token for EVERY call (wasteful, slow) token_url = f"{base_url}/identity/oauth/token" params = { "grant_type": "client_credentials", "client_id": client_id, "client_secret": client_secret, } response = requests.get(token_url, params=params) # ❌ Should be POST! token_data = response.json() access_token = token_data["access_token"] # ❌ No validation # ❌ No error handling # ❌ No token caching # ❌ Using GET instead of POST for token endpoint # ❌ No rate limit handling return {"base_url": base_url, "access_token": access_token} # =================================================================== # ✅ GOOD — env-based config, token caching, auto-refresh, retries # =================================================================== class MarketoError(Exception): """Base exception for Marketo integration errors.""" def __init__(self, message: str, error_code: Optional[str] = None): super().__init__(message) self.error_code = error_code class MarketoAuthError(MarketoError): """Authentication failed or token invalid/expired.""" pass class MarketoRateLimitError(MarketoError): """Rate limit exceeded.""" def __init__(self, message: str, retry_after: Optional[int] = None): super().__init__(message) self.retry_after = retry_after class MarketoNotFoundError(MarketoError): """Resource not found.""" pass # Marketo API error codes (common ones) MARKETO_ERROR_CODES = { "601": "Access token invalid", "602": "Access token expired", "603": "Access denied", "604": "Request timed out", "605": "Request limit exceeded", "606": "Rate limit exceeded", "607": "Daily quota reached", "608": "API temporarily unavailable", "609": "Invalid subscription", "610": "Invalid target", "611": "System error", "701": "Lead not found", "702": "Multiple leads match", "703": "Invalid fields in request", "704": "Sync failed", "705": "Invalid operation for type", "706": "Invalid JSON", "707": "Invalid parameter value", "708": "Parameter missing", "709": "Unsupported operation", "710": "Invalid batch size", "711": "Invalid token", } @dataclass class MarketoTokenStore: """Stores and manages Marketo OAuth tokens. Marketo access tokens expire after 3600 seconds (1 hour). We refresh proactively before expiry. """ access_token: Optional[str] = None expires_at: Optional[datetime] = None # When access_token expires def is_access_token_valid(self) -> bool: """Check if access token is still valid (has > 120 seconds left). Marketo tokens last 3600s. We use a 2-minute buffer to be safe. """ if not self.access_token or not self.expires_at: return False now = datetime.now(timezone.utc) buffer = timedelta(seconds=120) return self.expires_at > (now + buffer) def set_token( self, access_token: str, expires_in_seconds: int = 3600, ) -> None: """Set token from OAuth response. Args: access_token: New access token expires_in_seconds: TTL in seconds (default 3600 = 1 hour) """ self.access_token = access_token self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in_seconds) @dataclass class MarketoConfig: """Marketo configuration from environment variables. Environment variables: MARKETO_CLIENT_ID: OAuth client ID (from Admin → LaunchPoint) MARKETO_CLIENT_SECRET: OAuth client secret MARKETO_BASE_URL: Base URL, e.g., https://123-ABC-456.mktorest.com MARKETO_MUNCHKIN_ID: Optional (derived from base URL if not set) """ # Required client_id: Optional[str] = None client_secret: Optional[str] = None base_url: Optional[str] = None # Derived/optional munchkin_id: Optional[str] = None # Request config timeout: float = 30.0 max_retries: int = 5 initial_retry_delay: float = 1.0 max_retry_delay: float = 60.0 @classmethod def from_env(cls) -> "MarketoConfig": """Load configuration from environment variables.""" def parse_float(env_var: str, default: float) -> float: val = os.environ.get(env_var) if val is None: return default try: return float(val) except ValueError: return default def parse_int(env_var: str, default: int) -> int: val = os.environ.get(env_var) if val is None: return default try: return int(val) except ValueError: return default base_url = os.environ.get("MARKETO_BASE_URL", "") # Derive Munchkin ID from base URL if available # URL format: https://123-ABC-456.mktorest.com munchkin_id = os.environ.get("MARKETO_MUNCHKIN_ID") if not munchkin_id and base_url: # Extract from hostname import re match = re.search(r"https?://([a-zA-Z0-9-]+)\.mktorest\.com", base_url) if match: munchkin_id = match.group(1) return cls( client_id=os.environ.get("MARKETO_CLIENT_ID"), client_secret=os.environ.get("MARKETO_CLIENT_SECRET"), base_url=base_url.rstrip("/") if base_url else None, munchkin_id=munchkin_id, timeout=parse_float("MARKETO_TIMEOUT", 30.0), max_retries=parse_int("MARKETO_MAX_RETRIES", 5), ) def is_enabled(self) -> bool: """Check if Marketo is configured.""" return bool( self.client_id and self.client_secret and self.base_url ) def validate(self) -> bool: """Validate configuration. Returns: True if valid Raises: ValueError: If invalid when enabled """ if not self.is_enabled(): logger.info("Marketo not configured") return True if not self.client_id: raise ValueError("MARKETO_CLIENT_ID is required") if not self.client_secret: raise ValueError("MARKETO_CLIENT_SECRET is required") if not self.base_url: raise ValueError("MARKETO_BASE_URL is required") if not self.base_url.startswith("http"): raise ValueError( f"MARKETO_BASE_URL should start with https://, got: {self.base_url}" ) return True class MarketoClient: """Production-grade Marketo REST API client with token caching and retry handling. Features: - Config from environment - Automatic OAuth token caching and refresh - Exponential backoff for rate limits - Pagination helpers - Unified error handling with Marketo error code mapping """ # Common Activity Type IDs ACTIVITY_VISIT_WEBPAGE = 1 ACTIVITY_FILL_OUT_FORM = 2 ACTIVITY_CLICK_EMAIL = 3 ACTIVITY_OPEN_EMAIL = 4 ACTIVITY_DELIVER_EMAIL = 6 ACTIVITY_UNSUBSCRIBE_EMAIL = 7 ACTIVITY_CLICK_LINK = 8 ACTIVITY_ADD_TO_NURTURE = 9 ACTIVITY_PROGRESS_STATUS = 10 ACTIVITY_CHANGE_DATA_VALUE = 13 ACTIVITY_MERGE_LEADS = 15 ACTIVITY_ADD_TO_CAMPAIGN = 17 ACTIVITY_REMOVE_FROM_CAMPAIGN = 18 ACTIVITY_EMAIL_BOUNCE = 20 ACTIVITY_EMAIL_BOUNCE_SOFT = 21 ACTIVITY_CLICK_SALES_EMAIL = 27 ACTIVITY_OPEN_SALES_EMAIL = 28 ACTIVITY_INTERESTING_MOMENT = 46 ACTIVITY_SCORE_CHANGED = 47 # Lead partition filter types FILTER_TYPE_ID = "id" FILTER_TYPE_EMAIL = "email" FILTER_TYPE_SFDCLEADID = "sfdcLeadId" FILTER_TYPE_SFDCCONTACTID = "sfdcContactId" FILTER_TYPE_SFDCACCOUNTID = "sfdcAccountId" FILTER_TYPE_COOKIE = "cookie" FILTER_TYPE_TWITTERID = "twitterId" FILTER_TYPE_LINKEDINID = "linkedinId" FILTER_TYPE_FACEBOOKID = "facebookId" def __init__(self, config: MarketoConfig) -> None: self._config = config self._token_store = MarketoTokenStore() self._session: Optional[requests.Session] = None @property def _session(self) -> requests.Session: """Lazy-initialized requests session.""" if self.__session is None: self.__session = requests.Session() return self.__session @_session.setter def _session(self, value: Optional[requests.Session]) -> None: self.__session = value def _refresh_access_token(self) -> str: """Get a new access token using OAuth 2.0 client_credentials grant. Marketo token endpoint: POST /identity/oauth/token Returns: New access token Raises: MarketoAuthError: If token request fails """ if not REQUESTS_AVAILABLE: raise ImportError("requests library required for Marketo API") self._config.validate() token_url = urljoin(self._config.base_url, "/identity/oauth/token") params = { "grant_type": "client_credentials", "client_id": self._config.client_id, "client_secret": self._config.client_secret, } try: # Note: Marketo's token endpoint accepts either GET or POST # POST is more secure for credentials response = self._session.post( token_url, params=params, timeout=self._config.timeout, ) response.raise_for_status() token_data = response.json() # Check for Marketo error response if token_data.get("success") is False: errors = token_data.get("errors", []) error_msg = errors[0].get("message", "Unknown error") if errors else "Token request failed" raise MarketoAuthError(f"Token request failed: {error_msg}") access_token = token_data.get("access_token") if not access_token: raise MarketoAuthError("Token response missing access_token") expires_in = token_data.get("expires_in", 3600) # Update token store self._token_store.set_token( access_token=access_token, expires_in_seconds=expires_in, ) logger.info( "Marketo access token refreshed, expires in %d seconds", expires_in ) return access_token except RequestException as e: raise MarketoAuthError(f"Failed to get Marketo access token: {e}") from e def _get_valid_access_token(self) -> str: """Get a valid access token, refreshing if needed. Returns: Valid access token """ if self._token_store.is_access_token_valid(): return self._token_store.access_token # Need to refresh logger.info("Marketo token expired or missing, refreshing...") return self._refresh_access_token() def _build_url(self, endpoint: str) -> str: """Build full API URL from endpoint path. Args: endpoint: API endpoint path (e.g., "/rest/v1/leads.json") Returns: Full URL """ if endpoint.startswith("http"): return endpoint if not endpoint.startswith("/"): endpoint = "/" + endpoint return urljoin(self._config.base_url, endpoint) def _execute_with_retry( self, operation: Callable[[], requests.Response], operation_name: str = "operation", ) -> Dict[str, Any]: """Execute a Marketo API operation with retry for rate limits. Marketo error codes indicating retry: - 602: Access token expired (refresh and retry) - 606: Rate limit exceeded - 605: Request limit exceeded - 604: Request timed out Args: operation: Callable that returns requests Response operation_name: Name for logging Returns: Parsed JSON response dict Raises: MarketoError: Various error types based on response """ import random delay = self._config.initial_retry_delay last_exception: Optional[Exception] = None for attempt in range(self._config.max_retries): try: response = operation() # Parse JSON response try: data = response.json() except json.JSONDecodeError as e: raise MarketoError(f"Invalid JSON response: {e}") from e # Check for Marketo API errors in response body success = data.get("success", True) if not success: errors = data.get("errors", []) if errors: err = errors[0] error_code = str(err.get("code", "")) error_msg = err.get("message", "Unknown error") # Token expired - refresh and retry (once) if error_code == "602" and attempt == 0: logger.warning( "Marketo token expired (602), refreshing and retrying..." ) self._refresh_access_token() continue # Rate limit exceeded if error_code in ("604", "605", "606"): last_exception = MarketoRateLimitError( f"Marketo rate limit error ({error_code}): {error_msg}" ) wait_time = min( delay * (2 ** attempt) + random.uniform(0, 1), self._config.max_retry_delay ) logger.warning( "Marketo rate limited on %s (attempt %d/%d). Waiting %.1fs", operation_name, attempt + 1, self._config.max_retries, wait_time ) time.sleep(wait_time) continue # Access token invalid if error_code == "601": # Refresh once on first attempt if attempt == 0: logger.warning( "Marketo token invalid (601), refreshing..." ) self._refresh_access_token() continue raise MarketoAuthError( f"Marketo auth error ({error_code}): {error_msg}", error_code=error_code ) # Not found if error_code == "701": raise MarketoNotFoundError( f"Marketo resource not found: {error_msg}", error_code=error_code ) # Other errors raise MarketoError( f"Marketo API error ({error_code}): {error_msg}", error_code=error_code ) raise MarketoError("Marketo API request failed with no specific error") # Success return data except RequestException as e: # Network/HTTP errors last_exception = e if attempt < self._config.max_retries - 1: wait_time = min( delay * (2 ** attempt) + random.uniform(0, 1), self._config.max_retry_delay ) logger.warning( "Request error on %s (attempt %d/%d): %s. Retrying in %.1fs", operation_name, attempt + 1, self._config.max_retries, e, wait_time ) time.sleep(wait_time) continue raise MarketoError(f"Marketo request failed: {e}") from e # All retries exhausted raise MarketoRateLimitError( f"Rate limit retries exhausted after {self._config.max_retries} attempts" ) from last_exception def _request( self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None, json: Optional[Dict[str, Any]] = None, data: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None, operation_name: Optional[str] = None, ) -> Dict[str, Any]: """Make an authenticated API request to Marketo. Args: method: HTTP method (GET, POST, PUT, DELETE) endpoint: API endpoint path params: Query parameters json: JSON body (POST/PUT) data: Form data body headers: Additional headers operation_name: Name for logging Returns: Parsed response dict """ url = self._build_url(endpoint) op_name = operation_name or f"{method} {endpoint}" def _do_request() -> requests.Response: # Get fresh token each time (handles caching internally) access_token = self._get_valid_access_token() request_headers = { "Authorization": f"Bearer {access_token}", "Accept": "application/json", } if json is not None: request_headers["Content-Type"] = "application/json" if headers: request_headers.update(headers) return self._session.request( method=method, url=url, params=params, json=json, data=data, headers=request_headers, timeout=self._config.timeout, ) return self._execute_with_retry(_do_request, op_name) # =================================================================== # Lead Operations # =================================================================== def get_lead_by_id( self, lead_id: int, fields: Optional[List[str]] = None, ) -> Optional[Dict[str, Any]]: """Get a single Lead by ID. Args: lead_id: Marketo lead ID fields: List of field names to return (default: common fields) Returns: Lead dict or None if not found """ default_fields = [ "id", "email", "firstName", "lastName", "phone", "company", "createdAt", "updatedAt", "leadScore", "sfdcLeadId", "sfdcContactId" ] field_list = fields if fields else default_fields fields_param = ",".join(field_list) try: data = self._request( "GET", f"/rest/v1/leads/{lead_id}.json", params={"fields": fields_param}, operation_name=f"get_lead({lead_id})", ) results = data.get("result", []) if not results: return None return results[0] except MarketoNotFoundError: return None def get_leads_by_email( self, emails: List[str], fields: Optional[List[str]] = None, ) -> List[Dict[str, Any]]: """Get Leads by email addresses. Args: emails: List of email addresses (max 300 per request) fields: Fields to return Returns: List of lead dicts (empty if none found) """ if not emails: return [] default_fields = ["id", "email", "firstName", "lastName", "phone", "company"] field_list = fields if fields else default_fields # Marketo limit: 300 filter values per request all_results: List[Dict[str, Any]] = [] batch_size = 300 for i in range(0, len(emails), batch_size): batch = emails[i:i + batch_size] emails_param = ",".join(email.lower().strip() for email in batch) data = self._request( "GET", "/rest/v1/leads.json", params={ "filterType": "email", "filterValues": emails_param, "fields": ",".join(field_list), }, operation_name=f"get_leads_by_email[{i}:{i+batch_size}]", ) results = data.get("result", []) all_results.extend(results) return all_results def create_or_update_leads( self, leads: List[Dict[str, Any]], lookup_field: str = "email", action: str = "createOrUpdate", partition_name: Optional[str] = None, ) -> Dict[str, Any]: """Create or update Leads in batch. Actions: - createOrUpdate: Create if not exists, update if exists (default) - createOnly: Only create new, fail on duplicates - updateOnly: Only update existing, fail on missing - createDuplicate: Always create new (even if duplicate email) Args: leads: List of lead property dicts lookup_field: Field to use for deduplication (email, id, sfdcLeadId, etc.) action: Operation type partition_name: Optional lead partition name Returns: Response dict with 'result' list """ if not leads: return {"result": []} # Marketo batch limit: 300 leads per request batch_size = 300 all_results: List[Dict[str, Any]] = [] for i in range(0, len(leads), batch_size): batch = leads[i:i + batch_size] params = { "action": action, "lookupField": lookup_field, } if partition_name: params["partitionName"] = partition_name data = self._request( "POST", "/rest/v1/leads.json", params=params, json={"input": batch}, operation_name=f"create_or_update_leads[{i}:{i+batch_size}]", ) results = data.get("result", []) all_results.extend(results) logger.info( "Lead batch %d-%d: %d records processed", i, i + len(batch) - 1, len(results) ) return {"result": all_results} def delete_leads( self, lead_ids: List[int], ) -> Dict[str, Any]: """Delete Leads by ID. Args: lead_ids: List of lead IDs to delete (max 300 per request) Returns: Response dict """ if not lead_ids: return {"result": []} batch_size = 300 all_results: List[Dict[str, Any]] = [] for i in range(0, len(lead_ids), batch_size): batch = lead_ids[i:i + batch_size] data = self._request( "POST", "/rest/v1/leads/delete.json", json={"input": [{"id": lid} for lid in batch]}, operation_name=f"delete_leads[{i}:{i+batch_size}]", ) results = data.get("result", []) all_results.extend(results) return {"result": all_results} def describe_lead( self, ) -> Dict[str, Any]: """Get Lead object metadata (all fields and their types). Useful for discovering available fields and their properties. Returns: Describe response with 'result' containing field definitions """ return self._request( "GET", "/rest/v1/leads/describe.json", operation_name="describe_lead", ) # =================================================================== # Activities # =================================================================== def get_activity_types( self, ) -> List[Dict[str, Any]]: """Get all available activity type definitions. Returns: List of activity type dicts with id, name, description, attributes """ data = self._request( "GET", "/rest/v1/activities/types.json", operation_name="get_activity_types", ) return data.get("result", []) def get_activities( self, activity_type_ids: List[int], since_datetime: Optional[datetime] = None, lead_ids: Optional[List[int]] = None, batch_size: int = 300, max_total: Optional[int] = None, ) -> Generator[Dict[str, Any], None, None]: """Get activities, yielding them as they're fetched. Handles pagination automatically using nextPageToken. Args: activity_type_ids: List of activity type IDs to filter by since_datetime: Only get activities since this time lead_ids: Optional filter by lead IDs (max 30) batch_size: Page size (max 300) max_total: Optional maximum total activities to return Yields: Individual activity dicts """ if not activity_type_ids: return params: Dict[str, Any] = { "activityTypeIds": ",".join(str(t) for t in activity_type_ids), "batchSize": min(batch_size, 300), } if since_datetime: # Marketo uses ISO 8601 format params["sinceDatetime"] = since_datetime.strftime("%Y-%m-%dT%H:%M:%SZ") if lead_ids: # Marketo limit: 30 lead IDs for activity filter if len(lead_ids) > 30: logger.warning("lead_ids truncated to 30 (Marketo limit for activity filter)") lead_ids = lead_ids[:30] params["leadIds"] = ",".join(str(lid) for lid in lead_ids) next_page_token: Optional[str] = None total_yielded = 0 while True: if max_total and total_yielded >= max_total: break current_params = dict(params) if next_page_token: current_params["nextPageToken"] = next_page_token data = self._request( "GET", "/rest/v1/activities.json", params=current_params, operation_name="get_activities", ) results = data.get("result", []) if not results: break for activity in results: if max_total and total_yielded >= max_total: break yield activity total_yielded += 1 # Check for more pages next_page_token = data.get("nextPageToken") if not next_page_token: break logger.debug("Fetching next activity page with token: %s", next_page_token[:20] + "...") def get_lead_activities( self, lead_id: int, activity_type_ids: Optional[List[int]] = None, since_days: int = 30, ) -> List[Dict[str, Any]]: """Get recent activities for a specific Lead. Args: lead_id: Lead ID activity_type_ids: Optional filter by activity types since_days: How many days back to look Returns: List of activity dicts """ since_dt = datetime.now(timezone.utc) - timedelta(days=since_days) # Default activity types if not specified if activity_type_ids is None: activity_type_ids = [ self.ACTIVITY_VISIT_WEBPAGE, self.ACTIVITY_FILL_OUT_FORM, self.ACTIVITY_CLICK_EMAIL, self.ACTIVITY_OPEN_EMAIL, self.ACTIVITY_CHANGE_DATA_VALUE, self.ACTIVITY_SCORE_CHANGED, ] activities = list(self.get_activities( activity_type_ids=activity_type_ids, since_datetime=since_dt, lead_ids=[lead_id], )) return activities # =================================================================== # Campaigns # =================================================================== def get_campaigns( self, program_id: Optional[int] = None, is_triggerable: Optional[bool] = None, ) -> List[Dict[str, Any]]: """Get Smart Campaigns. Args: program_id: Optional filter by program ID is_triggerable: Optional filter for trigger campaigns (vs batch) Returns: List of campaign dicts """ params: Dict[str, Any] = {} if program_id is not None: params["programId"] = program_id data = self._request( "GET", "/rest/v1/campaigns.json", params=params, operation_name="get_campaigns", ) campaigns = data.get("result", []) if is_triggerable is not None: campaigns = [c for c in campaigns if c.get("isTriggerable", False) == is_triggerable] return campaigns def trigger_campaign( self, campaign_id: int, lead_ids: List[int], tokens: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, Any]: """Trigger a Smart Campaign for one or more Leads. The campaign must be a "Trigger Campaign" type (isTriggerable = true), not a Batch Campaign. Args: campaign_id: Smart Campaign ID lead_ids: List of lead IDs (max 100) tokens: Optional my tokens to override (e.g., [{"name": "{{my.Message}}", "value": "Hello"}]) Returns: Response dict """ if not lead_ids: raise ValueError("lead_ids is required") # Marketo limit: 100 leads per trigger call if len(lead_ids) > 100: logger.warning("lead_ids truncated to 100 (Marketo limit for trigger)") lead_ids = lead_ids[:100] payload: Dict[str, Any] = { "input": [{"id": lead_id} for lead_id in lead_ids] } if tokens: payload["tokens"] = tokens return self._request( "POST", f"/rest/v1/campaigns/{campaign_id}/trigger.json", json=payload, operation_name=f"trigger_campaign({c …(truncated)