# Directory Structure
```
├── .gitignore
├── LICENSE
├── mailchimp_mcp_client.py
├── mailchimp_mcp_server.py
├── mcp_overview.md
└── README.md
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# MailchimpMCP
Some utilities for developing an MCP server for the Mailchimp API
```
--------------------------------------------------------------------------------
/mailchimp_mcp_client.py:
--------------------------------------------------------------------------------
```python
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def main():
# Define how to start the MCP server (assuming it's in a local Python file)
server_params = StdioServerParameters(
command="python",
args=["mailchimp_mcp_server.py"]
)
# Launch the server and establish an MCP stdio connection
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 1. Initialize the MCP session (handshake)
await session.initialize()
print("MCP session initialized with Mailchimp server.")
# 2. List available tools provided by the Mailchimp MCP server
tools_response = await session.request({"method": "tools/list"})
tools_list = tools_response.get("tools", [])
print(f"Tools exposed by server: {[tool['name'] for tool in tools_list]}")
# (Each tool dict in tools_list has 'name', 'description', and an 'inputSchema')
# 3. Call a tool: list all campaigns
result = await session.request({
"method": "tools/call",
"params": {
"name": "list_campaigns",
"params": {} # no parameters required for this tool
}
})
campaigns = result # this should be the list of campaigns returned by our tool
print(f"\nRetrieved {len(campaigns)} campaigns from Mailchimp:")
for camp in campaigns:
print(f" - ID: {camp['id']}, Name: {camp['name']}, Status: {camp['status']}")
# 4. (Optional) Example of calling an action tool, like sending a campaign.
# Let's assume we have a campaign ID from the list to send.
if campaigns:
test_campaign_id = campaigns[0]['id'] # take the first campaign for demo
send_result = await session.request({
"method": "tools/call",
"params": {
"name": "send_campaign",
"params": {"campaign_id": test_campaign_id}
}
})
print(f"\nTool send_campaign result: {send_result}")
# 5. (Optional) List automations and start one
automations = await session.request({
"method": "tools/call",
"params": {"name": "list_automations", "params": {}}
})
print(f"\nFound {len(automations)} automation workflows.")
if automations:
workflow_id = automations[0]['id']
start_msg = await session.request({
"method": "tools/call",
"params": {"name": "start_automation", "params": {"workflow_id": workflow_id}}
})
print(f"start_automation result: {start_msg}")
# After this, the session will auto-close when exiting the context managers.
# Run the async main function
asyncio.run(main())
```
--------------------------------------------------------------------------------
/mailchimp_mcp_server.py:
--------------------------------------------------------------------------------
```python
import os
import requests
from mcp.server.fastmcp import FastMCP
# Initialize the MCP server with a descriptive name and optional version
mcp = FastMCP("MailchimpServer")
# --- Configuration & Authentication ---
# Fetch Mailchimp API credentials from environment (for security, avoid hardcoding)
MAILCHIMP_API_KEY = os.environ.get("MAILCHIMP_API_KEY", "<YOUR_API_KEY>")
MAILCHIMP_DC = os.environ.get("MAILCHIMP_DC", "<YOUR_DC>")
# The data center (DC) is typically the substring after the '-' in the API key, e.g. "us21"
# If not set explicitly, try to derive it from the API key
if MAILCHIMP_DC == "<YOUR_DC>" and MAILCHIMP_API_KEY and "-" in MAILCHIMP_API_KEY:
MAILCHIMP_DC = MAILCHIMP_API_KEY.split('-')[-1]
# Base URL for Mailchimp Marketing API
BASE_URL = f"https://{MAILCHIMP_DC}.api.mailchimp.com/3.0"
# A helper to perform Mailchimp API requests with proper authentication
def mailchimp_request(method: str, endpoint: str, **kwargs):
"""Make an HTTP request to Mailchimp API and return the response object."""
url = BASE_URL + endpoint
# Mailchimp uses HTTP Basic auth where username can be anything and password is the API key
auth = ("anystring", MAILCHIMP_API_KEY)
try:
response = requests.request(method, url, auth=auth, **kwargs)
except requests.RequestException as e:
# Network or connection error
raise Exception(f"Failed to connect to Mailchimp API: {e}")
# If the response status indicates an error, raise an exception with details
if response.status_code >= 400:
# Try to extract error message from Mailchimp's response JSON if available
error_detail = ""
try:
err_json = response.json()
# Mailchimp API errors often have keys like 'detail' or 'title' for error messages
error_detail = err_json.get("detail") or err_json.get("title") or str(err_json)
except ValueError:
error_detail = response.text or "Unknown error"
raise Exception(f"Mailchimp API error {response.status_code}: {error_detail}")
return response
# --- MCP Tool Definitions ---
@mcp.tool()
def list_campaigns() -> list:
"""Retrieve all email campaigns in the Mailchimp account (returns basic info for each campaign)."""
# Call Mailchimp API to list campaigns
resp = mailchimp_request("GET", "/campaigns")
data = resp.json()
campaigns = []
for camp in data.get("campaigns", []):
campaigns.append({
"id": camp.get("id"),
"name": camp.get("settings", {}).get("title") or camp.get("settings", {}).get("subject_line"),
"status": camp.get("status"),
"emails_sent": camp.get("emails_sent")
})
return campaigns
@mcp.tool()
def create_campaign(list_id: str, subject: str, from_name: str, reply_to: str) -> dict:
"""Create a new email campaign in Mailchimp (returns the new campaign's ID and details)."""
# Prepare the campaign payload (using 'regular' campaign type)
payload = {
"type": "regular",
"recipients": {"list_id": list_id},
"settings": {
"subject_line": subject,
"from_name": from_name,
"reply_to": reply_to
}
}
resp = mailchimp_request("POST", "/campaigns", json=payload)
campaign_info = resp.json()
# Return key details of the created campaign (id and status)
return {"id": campaign_info.get("id"), "status": campaign_info.get("status", "created")}
@mcp.tool()
def send_campaign(campaign_id: str) -> str:
"""Send a campaign that has been created (campaign must be ready to send)."""
# Hitting the send action endpoint for the specified campaign
mailchimp_request("POST", f"/campaigns/{campaign_id}/actions/send")
# If successful (no exception raised), Mailchimp will have queued/sent the campaign
return f"Campaign {campaign_id} has been sent."
@mcp.tool()
def list_automations() -> list:
"""List all classic automation workflows in the Mailchimp account."""
resp = mailchimp_request("GET", "/automations")
data = resp.json()
automations = []
for auto in data.get("automations", []):
automations.append({
"id": auto.get("id"),
"name": auto.get("settings", {}).get("title") or auto.get("create_time"), # title if present
"status": auto.get("status"),
"emails_sent": auto.get("emails_sent")
})
return automations
@mcp.tool()
def start_automation(workflow_id: str) -> str:
"""Start all emails in a specified automation workflow (activating the automation)."""
mailchimp_request("POST", f"/automations/{workflow_id}/actions/start-all-emails")
return f"Automation workflow {workflow_id} started."
# We could add more tools for other operations (pause automation, add subscribers, etc.) following the same pattern.
if __name__ == "__main__":
# Run the MCP server. This will listen for incoming MCP client connections (stdio by default).
print("Starting Mailchimp MCP server... (press Ctrl+C to stop)")
mcp.run()
```
--------------------------------------------------------------------------------
/mcp_overview.md:
--------------------------------------------------------------------------------
```markdown
# Overview of MCP (Model Context Protocol)
Anthropic’s **Model Context Protocol (MCP)** is an open standard designed to bridge AI assistants (LLMs) with external data sources and tools. The core idea is to provide a **universal interface** – often analogized to a “USB-C port” for AI – so that an LLM-based application can plug into various databases, APIs, file systems, or services in a consistent way. This addresses the problem of LLMs being “trapped” in isolation, unable to access up-to-date or proprietary information without bespoke integrations for each source. By replacing fragmented one-off connectors with a single standardized protocol, MCP simplifies development and ensures AI systems can retrieve relevant context and perform actions using any MCP-compatible source.
MCP’s design philosophy emphasizes:
- **Separation of Concerns:** Distinguishing data access from AI reasoning.
- **Interoperability:** Allowing developers to expose data or actions via MCP servers, and letting AI applications use a common MCP client to leverage them.
> **Figure:** *Overview of the MCP architecture – an AI application (MCP host, e.g. Claude or an IDE) connects via MCP clients to multiple MCP servers that bridge to various data sources. Each MCP server exposes a specific domain (e.g. Slack, Gmail, Calendar, or local files) through the standard protocol, allowing the LLM to query or act on those resources.*
---
# How MCP Works (Architecture & Workflows)
## Core Workflow
- **Establish Connection:** An MCP client establishes a connection to an MCP server.
- **Initialization Handshake:** The client sends an `initialize` request (with protocol version and capabilities), and the server responds with its own capabilities. Once agreed upon, an `initialized` notification is exchanged.
- **Message Exchange:** Communication proceeds via JSON-formatted messages (requests, responses, and notifications) following the JSON-RPC 2.0 standard.
- **Termination:** Either side can gracefully close the connection when the session is complete.
## Transport Mechanisms
- **STDIO:** The server reads JSON messages from `stdin` and writes responses to `stdout`. This is ideal for local integrations.
- **HTTP + SSE (Server-Sent Events):** In remote setups, the client sends HTTP `POST` requests and the server uses SSE to push responses and notifications back.
## Message Types
- **Requests:** Calls that expect a result (with a method name, parameters, and an ID).
- **Results:** Successful responses containing the output data.
- **Errors:** Responses indicating failures, with standard error codes.
- **Notifications:** One-way messages for events or updates that do not expect a reply.
## Components for Context Management
- **Resources:** Read-only data (e.g., documents, database entries) that the AI can pull into its context.
- **Tools:** Actions or functions (e.g., sending an email) that the AI can invoke to effect changes.
- **Prompts:** Pre-defined prompt templates or workflows to guide the AI’s interactions.
A typical example involves the AI requesting a resource (like a log file) from a server, using the returned content to generate a summary, or invoking a tool to send an email. The MCP client abstracts the discovery, invocation, and data transfer steps.
---
# Design Considerations for MCP Implementations
## Message Formatting & Protocol Compliance
- **JSON-RPC 2.0:** All MCP messages must adhere to this standard.
- **Method Naming:** Use standard method names (e.g., `initialize`, `tools/list`, `tools/call`, `resources/list`, `resources/read`, `prompts/list`).
- **JSON Schema:** Define schemas for tool inputs and resource formats to validate incoming requests.
## Request & Response Workflow
- **Initialization:** Advertise server capabilities during the handshake.
- **Idempotency:** Ensure safe methods (e.g., status queries) are idempotent.
- **Streaming vs. Atomic Responses:** Consider breaking large responses into chunks or sending progress notifications for long operations.
## Error Handling
- **Standard Codes:** Use JSON-RPC error codes (like -32601 for “Method not found” or -32602 for “Invalid params”).
- **Input Validation:** Validate parameters rigorously to avoid processing errors.
- **Exception Handling:** Convert exceptions into structured MCP error responses.
## Scalability & Performance
- **Concurrency:** Use asynchronous programming or multi-threading to handle multiple requests.
- **Transport Impact:** Choose the appropriate transport (local STDIO vs. remote HTTP+SSE) based on expected usage.
- **Caching & Rate Limiting:** Implement caching for frequently requested data and rate limiting to prevent overload.
## Security Best Practices
- **Transport Security:** Use TLS/HTTPS for remote connections.
- **Authentication & Authorization:** Enforce API keys or OAuth tokens and validate each request.
- **Input Sanitization:** Prevent directory traversal, injection attacks, or other malicious inputs.
- **Access Control:** Limit sensitive operations to authorized clients.
- **Audit & Monitoring:** Log important actions and monitor for abuse or anomalies.
## Additional Tips
- **Leverage SDKs:** Utilize MCP SDKs to simplify type safety, validation, and development.
- **Testing:** Implement thorough unit tests for both MCP clients and servers.
- **Documentation:** Clearly document the capabilities of your MCP server.
- **Human Oversight:** Ensure that any high-impact tool calls require explicit user confirmation.
---
# Technical Deep Dive: MCP Protocol Architecture & Context Management
## Protocol Architecture and Lifecycle
MCP is a specialized RPC layer built on JSON-RPC 2.0. It abstracts the networking details by providing classes (like `Protocol`, `Client`, and `Server`) that manage:
- Correlation of requests and responses.
- Asynchronous message handling.
- Callback management for incoming requests.
When a client calls a method (e.g., `tools/call`), the SDK handles packaging the request, sending it over the transport, waiting for the response, and then decoding it back into a native Python object.
## Transport Layer Details
- **STDIO Transport:**
Uses subprocess I/O streams with a framing mechanism (newline-delimited or length-prefixed JSON) for message boundaries.
- **HTTP + SSE Transport:**
The server runs an HTTP endpoint; clients send POST requests while the server pushes responses and notifications via SSE. This setup is ideal for remote deployments.
## Context Propagation Mechanisms
- **Flow:** Context data (like document contents) flows from the MCP server to the client and is then incorporated into the LLM’s prompt.
- **Client Responsibility:** The MCP client decides when and how to fetch and integrate context (e.g., based on user selection or automated rules).
- **Structured Data:** MCP delivers context with metadata (e.g., MIME type, URI) to help the client make informed decisions.
## State Management
- **Stateful vs. Stateless:**
Servers can be designed to maintain session state (e.g., authentication tokens, database connections) or require all necessary information with each request.
- **Session Initialization:**
The initialization handshake can set up session-specific context, such as storing API keys securely.
## Prompts and Automation
- **Prompt Templates:**
MCP can deliver pre-defined prompts that guide the LLM through complex workflows.
- **Workflow Integration:**
Tools can be chained together within a prompt to automate multi-step processes, like creating a campaign or starting an automation.
## Extensibility and Modular Design
- **Single-Responsibility Servers:**
Each MCP server should focus on a specific domain (e.g., Mailchimp, Slack, or Google Analytics).
- **Multiple Connections:**
The MCP client can manage multiple server connections concurrently, enabling the AI to integrate various services seamlessly.
- **Dynamic Discovery:**
The client can query available tools or resources, allowing the AI to adapt its behavior based on what’s available.
## Performance Considerations
- **Serialization Overhead:**
JSON serialization and deserialization can add latency, especially for large payloads. Use chunking or resource URIs for very large data.
- **Parallel Calls:**
Design the system to handle concurrent MCP calls where possible, minimizing the round-trip latency.
- **Caching:**
Cache frequently requested context to reduce redundant API calls.
## Error Propagation and Context Handling
- **Structured Errors:**
Errors are returned with standardized codes and messages, allowing the client to decide whether to expose them to the LLM.
- **User Feedback:**
Determine if errors should be visible to the end-user or handled silently by the client.
- **Adaptive Behavior:**
Use error responses to adjust subsequent calls (e.g., prompting for reauthentication or alternative actions).
## Closing the Loop
- **Integrating External Data:**
The final AI response can reference external context (e.g., “I retrieved the campaign details from Mailchimp…”), increasing transparency.
- **User Trust:**
Informing users about the external sources of context helps build trust in the AI’s actions.
---
# Example Implementation: MCP Server and Client for Mailchimp Marketing API
This example demonstrates a complete Python implementation of an MCP server and client that integrates with the Mailchimp Marketing API. The implementation supports advanced functionality such as campaign management and automation workflows, and it includes proper authentication.
## MCP Server Implementation (`mailchimp_mcp_server.py`)
```python
import os
import requests
from mcp.server.fastmcp import FastMCP
# Initialize the MCP server with a descriptive name and optional version
mcp = FastMCP("MailchimpServer")
# --- Configuration & Authentication ---
# Fetch Mailchimp API credentials from environment (for security, avoid hardcoding)
MAILCHIMP_API_KEY = os.environ.get("MAILCHIMP_API_KEY", "<YOUR_API_KEY>")
MAILCHIMP_DC = os.environ.get("MAILCHIMP_DC", "<YOUR_DC>")
# Derive data center from API key if not explicitly provided
if MAILCHIMP_DC == "<YOUR_DC>" and MAILCHIMP_API_KEY and "-" in MAILCHIMP_API_KEY:
MAILCHIMP_DC = MAILCHIMP_API_KEY.split('-')[-1]
# Base URL for Mailchimp Marketing API
BASE_URL = f"https://{MAILCHIMP_DC}.api.mailchimp.com/3.0"
# A helper to perform Mailchimp API requests with proper authentication
def mailchimp_request(method: str, endpoint: str, **kwargs):
"""Make an HTTP request to Mailchimp API and return the response object."""
url = BASE_URL + endpoint
# Mailchimp uses HTTP Basic auth where username can be anything and password is the API key
auth = ("anystring", MAILCHIMP_API_KEY)
try:
response = requests.request(method, url, auth=auth, **kwargs)
except requests.RequestException as e:
raise Exception(f"Failed to connect to Mailchimp API: {e}")
if response.status_code >= 400:
error_detail = ""
try:
err_json = response.json()
error_detail = err_json.get("detail") or err_json.get("title") or str(err_json)
except ValueError:
error_detail = response.text or "Unknown error"
raise Exception(f"Mailchimp API error {response.status_code}: {error_detail}")
return response
# --- MCP Tool Definitions ---
@mcp.tool()
def list_campaigns() -> list:
"""Retrieve all email campaigns in the Mailchimp account."""
resp = mailchimp_request("GET", "/campaigns")
data = resp.json()
campaigns = []
for camp in data.get("campaigns", []):
campaigns.append({
"id": camp.get("id"),
"name": camp.get("settings", {}).get("title") or camp.get("settings", {}).get("subject_line"),
"status": camp.get("status"),
"emails_sent": camp.get("emails_sent")
})
return campaigns
@mcp.tool()
def create_campaign(list_id: str, subject: str, from_name: str, reply_to: str) -> dict:
"""Create a new email campaign in Mailchimp."""
payload = {
"type": "regular",
"recipients": {"list_id": list_id},
"settings": {
"subject_line": subject,
"from_name": from_name,
"reply_to": reply_to
}
}
resp = mailchimp_request("POST", "/campaigns", json=payload)
campaign_info = resp.json()
return {"id": campaign_info.get("id"), "status": campaign_info.get("status", "created")}
@mcp.tool()
def send_campaign(campaign_id: str) -> str:
"""Send a campaign that has been created."""
mailchimp_request("POST", f"/campaigns/{campaign_id}/actions/send")
return f"Campaign {campaign_id} has been sent."
@mcp.tool()
def list_automations() -> list:
"""List all automation workflows in Mailchimp."""
resp = mailchimp_request("GET", "/automations")
data = resp.json()
automations = []
for auto in data.get("automations", []):
automations.append({
"id": auto.get("id"),
"name": auto.get("settings", {}).get("title") or auto.get("create_time"),
"status": auto.get("status"),
"emails_sent": auto.get("emails_sent")
})
return automations
@mcp.tool()
def start_automation(workflow_id: str) -> str:
"""Start all emails in a specified automation workflow."""
mailchimp_request("POST", f"/automations/{workflow_id}/actions/start-all-emails")
return f"Automation workflow {workflow_id} started."
if __name__ == "__main__":
print("Starting Mailchimp MCP server... (press Ctrl+C to stop)")
mcp.run()
```