# Directory Structure
```
├── .gitignore
├── .python-version
├── LICENSE
├── main.py
├── pdm.lock
├── pyproject.toml
├── README.md
├── src
│ ├── test.py
│ └── titan_mind
│ ├── __init__.py
│ ├── constants.py
│ ├── networking.py
│ ├── server.py
│ ├── titan_mind_functions.py
│ └── utils
│ ├── __init__.py
│ ├── app_specific
│ │ ├── __init__.py
│ │ ├── mcp.py
│ │ ├── networking
│ │ │ ├── __init__.py
│ │ │ └── titan_mind.py
│ │ └── utils.py
│ └── general
│ ├── __init__.py
│ ├── date_time.py
│ ├── mcp.py
│ └── networking.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
1 | 3.11
2 |
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | .pdm-python
2 | .venv/
3 | __pypackages__/
4 | # Ignore distribution archives
5 | dist.zip
6 |
7 | # Ignore IntelliJ / PyCharm project files
8 | .idea/
9 |
10 | # Ignore the build distribution directory
11 | dist/
12 |
13 | .env
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | ## Titanmind WhatsApp MCP
2 |
3 | A WhatsApp marketing and messaging tool MCP (Model Control Protocol) service using [Titanmind](https://www.titanmind.so/). Handles free-form messages (24hr window) and template workflows automatically
4 |
5 | ## Overview
6 |
7 | This service provides all the WhatsApp marketing and messaging functionalities using Titanmind. Includes features like template creation and registration with all components header, body, CTAs.., template broadcast to phone numbers in bulk. Read and send messages in an active conversation.
8 |
9 | > This MCP utilizes Titanmind. Titanmind Account is a requirement to use this MCP.
10 | >
11 | > Titanmind enhances WhatsApp communication by providing powerful features such as **conversation management, scheduling, agentic conversations, content generation etc.**
12 |
13 | ## Features
14 |
15 | #### Conversation Management
16 |
17 | **Get Recent Conversations**
18 |
19 | * Retrieve all conversations with messages sent or received in the last 24 hours
20 | * Returns conversation data with recent activity
21 |
22 | **Get Conversation Messages**
23 |
24 | * Fetch all messages from a specific conversation
25 | * Requires: `conversation_id` (alphanumeric conversation identifier)
26 |
27 | **Send WhatsApp Message**
28 |
29 | * Send a message to an existing WhatsApp conversation
30 | * Requires: `conversation_id` and `message` content
31 |
32 | #### Template Management
33 |
34 | **Create Message Template**
35 |
36 | * Register new WhatsApp message templates for approval
37 | * Configure template name (single word, underscores allowed only)
38 | * Set language (default: "en") and category (MARKETING, UTILITY, AUTHENTICATION)
39 | * Structure message components including:
40 | * **BODY** (required): Main text content
41 | * **HEADER** (optional): TEXT, VIDEO, IMAGE, or DOCUMENT format
42 | * **FOOTER** (optional): Footer text
43 | * **BUTTONS** (optional): QUICK\_REPLY, URL, or PHONE\_NUMBER actions
44 |
45 | **Get Templates**
46 |
47 | * Retrieve all created templates with approval status
48 | * Optional filtering by template name
49 |
50 | **Send Bulk Messages**
51 |
52 | * Send messages to multiple phone numbers using approved templates
53 | * Requires: `template_id` and list of contacts
54 | * Contact format: country code alpha (e.g., "IN"), country code (e.g., "91"), and phone number
55 |
56 | ## Installation
57 |
58 | ### Prerequisites
59 |
60 | * Python 3.10 or higher
61 | * API Key and Business Code from [Titanmind](https://www.titanmind.so/)
62 |
63 | ### Usage with MCP Client
64 |
65 | In any MCP Client like Claude or Cursor, Titanmind whatsapp MCP config can be added following ways:
66 |
67 | #### Using [Titanmind WhatsApp MCP Python package](https://pypi.org/project/titanmind-whatsapp-mcp/0.1.2/)
68 | 1\. Install pipx to install the python package globally
69 |
70 | ```plaintext
71 | # terminal
72 |
73 | # Install pipx first
74 | brew install pipx # on macOS
75 | # or
76 | sudo apt install pipx # on Ubuntu/Debian
77 |
78 | # Then install Titanmind WhatsApp MCP Python package
79 | pipx install titanmind-whatsapp-mcp
80 |
81 | # Make sure '/[HOME_DIR_OR_USER_PRFILE]/.local/bin' is on your PATH environment variable. Use pipx ensurepath to set it.
82 | pipx ensurepath
83 |
84 | ```
85 |
86 | 2\. Set the MCP Config python package script in the MCP Client's MCP Configs Json file.
87 |
88 | ```plaintext
89 | {
90 | "mcpServers": {
91 | "TitanMindMCP": {
92 | "command": "/[HOME_DIR_OR_USER_PRFILE]/.local/bin/titan-mind-mcp",
93 | "args": [
94 | ],
95 | "env": {
96 | "api-key": "XXXXXXXXXXXXXXXXXXXXXXXX",
97 | "bus-code": "XXXXXX"
98 | }
99 | }
100 | }
101 | }
102 |
103 | ```
104 |
105 | #### Use Remote Titanmind MCP server config
106 |
107 | 1\. Make sure npx is installed in the system
108 | 2\. Then just add the MCP config
109 |
110 | ```plaintext
111 | {
112 | "mcpServers": {
113 | "TitanMindMCP": {
114 | "command": "npx",
115 | "args": [
116 | "mcp-remote",
117 | "https://mcp.titanmind.so/whatsapp/mcp/",
118 | "--header",
119 | "api-key:XXXXXXXXXXXXXXXXXXXXXXX",
120 | "--header",
121 | "bus-code:XXXXXX"
122 | ]
123 | }
124 | }
125 | }
126 | ```
127 |
128 | #### Use local python project config
129 |
130 | 1\. First Setup project using instructions mentioned in the Setup Project section.
131 | 2\. Then add the MCP config
132 |
133 | ```plaintext
134 | {
135 | "mcpServers": {
136 | "TitanMindMCP": {
137 | "type": "stdio",
138 | "command": "uv",
139 | "args": [
140 | "run",
141 | "--directory",
142 | "/[PATH_TO_THE_PROJECT]",
143 | "python",
144 | "main.py"
145 | ],
146 | "env": {
147 | "api-key": "XXXXXXXXXXXXXXXXXXXX",
148 | "bus-code": "XXXXXX"
149 | }
150 | }
151 | }
152 | }
153 | ```
154 |
155 | ### Manual Installation for custom purpose or development
156 |
157 | ### Install package from PyPI for package use
158 |
159 | ```plaintext
160 | pip install titanmind-whatsapp-mcp
161 | ```
162 |
163 | Or use `uv`:
164 |
165 | ```plaintext
166 | uv pip install titanmind-whatsapp-mcp
167 | ```
168 |
169 | ### Setup Project for development use
170 |
171 | 1\. Clone the repository:
172 |
173 | ```plaintext
174 | git clone https://github.com/TitanmindAGI/titanmind-whatsapp-mcp
175 | cd titanmind-whatsapp-mcp
176 | ```
177 |
178 | 2\. Install dependencies:
179 |
180 | ```plaintext
181 | pip install -e .
182 | # Or
183 | uv pip install -e .
184 | ```
185 |
186 | 3\. Set the auth keys
187 |
188 | ```plaintext
189 | export api-key="your-titanmind-api-key"
190 | export bus-code="your-titanmind-business-code"
191 | ```
192 |
193 | ## How it Works
194 |
195 | TitanMind's WhatsApp messaging system operates under two distinct messaging modes based on timing and conversation status:
196 |
197 | ## Free-Form Messaging (24-Hour Window)
198 |
199 | * **When Available**: Only after a user has sent a message within the last 24 hours
200 | * **Content Freedom**: Any content is allowed without pre-approval
201 | * **Use Case**: Ongoing conversations and immediate responses
202 |
203 | ## Template Messaging (Outside 24-Hour Window)
204 |
205 | * **When Required**: For new conversations or when the 24-hour window has expired
206 | * **Content Structure**: Pre-approved, structured message templates only
207 | * **Use Case**: Initial outreach and re-engagement campaigns
208 |
209 | ## Messaging Workflow Process
210 |
211 | 1. **Check Messaging Window Status**
212 | * Verify if receiver's phone number is within the free-form messaging window
213 | * A receiver is eligible for free-form messaging if:
214 | * A conversation with their phone number already exists AND
215 | * The receiver has sent a message within the last 24 hours
216 | 2. **Choose Messaging Method**
217 | * **Free-Form**: Send directly if within 24-hour window
218 | * **Template**: Register and use approved template if outside window
219 | 3. **Template Approval Process** (if needed)
220 | * Submit template for WhatsApp approval
221 | * Wait for approval confirmation
222 | * Template becomes available for bulk messaging
223 | 4. **Send Message**
224 | * Execute message delivery using appropriate method
225 | * Monitor delivery status
226 | 5. **Verify Delivery**
227 | * Check conversation to confirm receiver successfully received the message
228 | * Track message status and engagement
229 |
230 | ## Usage Notes
231 |
232 | * All tools integrate with Titanmind's WhatsApp channel messaging functionality
233 | * Templates require approval before they can be used for bulk messaging
234 | * For more help contact us through [https://www.titanmind.so/](https://www.titanmind.so/)
235 |
236 | ## License
237 |
238 | MIT License - See LICENSE file
```
--------------------------------------------------------------------------------
/src/titan_mind/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/src/titan_mind/constants.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/app_specific/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/app_specific/networking/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/general/__init__.py:
--------------------------------------------------------------------------------
```python
1 |
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/general/networking.py:
--------------------------------------------------------------------------------
```python
1 |
2 |
3 |
```
--------------------------------------------------------------------------------
/main.py:
--------------------------------------------------------------------------------
```python
1 | from titan_mind.server import main as server_main
2 |
3 | def main():
4 | server_main()
5 |
6 |
7 | if __name__ == "__main__":
8 | main()
9 |
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/general/mcp.py:
--------------------------------------------------------------------------------
```python
1 | from fastmcp.server.dependencies import get_http_request
2 | from starlette.requests import Request as StarletteRequest
3 |
4 | def get_the_headers_from_the_current_mcp_request():
5 | request: StarletteRequest = get_http_request()
6 | return request.headers
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/app_specific/mcp.py:
--------------------------------------------------------------------------------
```python
1 | from titan_mind.utils.app_specific.utils import to_run_mcp_in_server_mode_or_std_io
2 | from titan_mind.utils.general.mcp import get_the_headers_from_the_current_mcp_request
3 | import os
4 |
5 |
6 | def get_the_api_key() -> str:
7 | if to_run_mcp_in_server_mode_or_std_io():
8 | api_key = get_the_headers_from_the_current_mcp_request().get("api-key")
9 | else:
10 | api_key = os.environ.get("api-key")
11 |
12 | return api_key
13 |
14 |
15 | def get_the_business_code() -> str:
16 | if to_run_mcp_in_server_mode_or_std_io():
17 | business_code = get_the_headers_from_the_current_mcp_request().get("bus-code")
18 | else:
19 | business_code = os.environ.get("bus-code")
20 |
21 | return business_code
22 |
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/general/date_time.py:
--------------------------------------------------------------------------------
```python
1 | import datetime
2 |
3 |
4 | def get_date_time_to_utc_server_time_format_string(date_time_inst: datetime.datetime) -> str:
5 | """
6 | Convert a datetime object to UTC format string: YYYY-MM-DDTHH:MM:SS.fffffZ
7 |
8 | Args:
9 | dt: datetime object (can be naive or timezone-aware)
10 |
11 | Returns:
12 | str: UTC formatted string ending with 'Z'
13 | """
14 | # If datetime is naive (no timezone info), assume it's UTC
15 | if date_time_inst.tzinfo is None:
16 | dt = date_time_inst.replace(tzinfo=datetime.timezone.utc)
17 |
18 | # Convert to UTC if it's in a different timezone
19 | dt_utc = date_time_inst.astimezone(datetime.timezone.utc)
20 |
21 | # Format to ISO string with microseconds and 'Z' suffix
22 | return dt_utc.strftime('%Y-%m-%dT%H:%M:%S.%fZ')
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/app_specific/utils.py:
--------------------------------------------------------------------------------
```python
1 | import argparse
2 | import os
3 |
4 | from dotenv import load_dotenv
5 |
6 | load_dotenv()
7 |
8 |
9 | def to_run_mcp_in_server_mode_or_std_io() -> bool:
10 | print(f"is_the_mcp_to_run_in_server_mode_or_std_dio: {get_script_args().run_in_server_mode}")
11 | return get_script_args().run_in_server_mode
12 |
13 |
14 | def get_script_args():
15 | parser = argparse.ArgumentParser(
16 | description="MCP server for WhatsApp functionality via Titanmind."
17 | )
18 |
19 | parser.add_argument(
20 | "--run-in-server-mode",
21 | action="store_true",
22 | # even of the presence of --run-in-server-mode will make set the bool as true. no need to provide True/False to this flag
23 | help="bool to run in server mode or stdio mode"
24 | )
25 |
26 | parser.add_argument(
27 | "--port",
28 | type=int,
29 | default=3000,
30 | help="The port number to run the server on (default: 3000)."
31 | )
32 |
33 | return parser.parse_args()
34 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [build-system]
2 | requires = ["pdm-backend>=2.4.0"]
3 | build-backend = "pdm.backend"
4 |
5 | # To upload to pypi
6 | # use ```rm -rf dist/``` to clean build folder,
7 | # use ```python -m build``` to build
8 | # and use ```twine upload dist/*``` to upload the build
9 |
10 | [project]
11 | name = "titanmind-whatsapp-mcp"
12 | version = "0.1.8"
13 | description = "MCP server for WhatsApp functionality via Titanmind"
14 | authors = [
15 | {name = "Shubhashish", email = "[email protected]"}
16 | ]
17 | readme = "README.md"
18 | requires-python = ">=3.10"
19 | dependencies = [
20 | "fastmcp>=2.0.0",
21 | "requests>=2.32.3",
22 | "pydantic>=2.5.0",
23 | "pydantic-core>=2.14.0",
24 | "requests_oauthlib>=2.0.0",
25 | "python-dotenv>=1.1.1"
26 | ]
27 |
28 | # command/script to run the whole package --> titan-mind-mcp. it directly runs the server file in src/titan_mind
29 | [project.scripts]
30 | titan-mind-mcp = "titan_mind.server:main"
31 |
32 | [tool.setuptools]
33 | packages = ["titan_mind"]
34 | package-dir = {"" = "src"}
35 |
36 |
37 | [tool.pdm]
38 | distribution = true
39 |
40 | license = {text = "MIT"}
41 | classifiers = [
42 | "Programming Language :: Python :: 3",
43 | "License :: OSI Approved :: MIT License",
44 | "Operating System :: OS Independent",
45 | ]
46 |
47 | [project.urls]
48 | "Homepage" = "https://github.com/TitanmindAGI/titan-mind-whatsapp-mcp"
49 | "Repository" = "https://github.com/TitanmindAGI/titan-mind-whatsapp-mcp"
50 |
51 |
```
--------------------------------------------------------------------------------
/src/titan_mind/networking.py:
--------------------------------------------------------------------------------
```python
1 | import json
2 |
3 | from titan_mind.utils.app_specific.mcp import get_the_api_key, get_the_business_code
4 |
5 | _titan_engage_base_base_url = 'https://api.titanmind.so'
6 | _titan_engage_base_url = f'{_titan_engage_base_base_url}/api/'
7 |
8 |
9 | def get_titan_engage_headers() -> dict:
10 | return {
11 | 'accept': 'application/json, text/plain, */*',
12 | 'content-type': 'application/json',
13 | 'authorization': f'API-Key {get_the_api_key()}',
14 | 'x-business-code': f'{get_the_business_code()}',
15 | }
16 |
17 |
18 | def get_titan_engage_url(endpoint: str):
19 | return f"{_titan_engage_base_url}{endpoint}"
20 |
21 |
22 | def print_request_and_response(response):
23 | """
24 | Accepts a requests.Response object and prints details of both the
25 | request that generated it and the response itself.
26 | Then, it returns the original response object.
27 |
28 | Args:
29 | response: The requests.Response object to print details for.
30 |
31 | Returns:
32 | The original requests.Response object.
33 | """
34 | # --- Print Request Details ---
35 | print("-" * 30)
36 | print("REQUEST SENT:")
37 | print("-" * 30)
38 | print(f"Method: {response.request.method}")
39 | print(f"URL: {response.request.url}")
40 | print("Headers:")
41 | for header, value in response.request.headers.items():
42 | print(f" {header}: {value}")
43 | if response.request.body:
44 | # Decode body for readability if it's bytes (e.g., from POST requests)
45 | try:
46 | print(f"Body: {response.request.body.decode('utf-8')}")
47 | except AttributeError:
48 | print(f"Body: {response.request.body}") # Already string or None
49 | else:
50 | print("Body: (No body for GET request)")
51 | print("\n")
52 |
53 | # --- Print Response Details ---
54 | print("-" * 30)
55 | print("RESPONSE RECEIVED:")
56 | print("-" * 30)
57 | print(f"Status Code: {response.status_code}")
58 | print(f"Reason: {response.reason}") # e.g., "OK", "Not Found"
59 | print("Headers:")
60 | for header, value in response.headers.items():
61 | print(f" {header}: {value}")
62 |
63 | print("\nBody (JSON/Text):")
64 | try:
65 | response_json = response.json()
66 | print(json.dumps(response_json, indent=2))
67 | except json.JSONDecodeError:
68 | print(response.text)
69 | print("\n")
70 |
71 | return response
72 |
```
--------------------------------------------------------------------------------
/src/test.py:
--------------------------------------------------------------------------------
```python
1 | #!/usr/bin/env python3
2 | """
3 | Test script for the WhatsApp MCP server
4 | """
5 |
6 | import asyncio
7 | import os
8 |
9 | from titan_mind.server import mcp
10 |
11 |
12 | async def test_tools():
13 | """Test the MCP server tools directly"""
14 | print("🧪 Testing WhatsApp MCP Server Tools\n")
15 |
16 | # Set up test environment
17 | os.environ.setdefault("TITANENGAGE_API_KEY", "test-key-for-testing")
18 |
19 | try:
20 | # Test 1: List available tools
21 | print("📋 Available Tools:")
22 | tools = mcp.list_tools()
23 | for tool in tools:
24 | print(f" - {tool.name}: {tool.description}")
25 | print()
26 |
27 | # Test 2: List available prompts
28 | print("💬 Available Prompts:")
29 | prompts = mcp.list_prompts()
30 | for prompt in prompts:
31 | print(f" - {prompt.name}: {prompt.description}")
32 | print()
33 |
34 | # Test 3: Test a prompt
35 | print("🎯 Testing WhatsApp Message Template Prompt:")
36 | try:
37 | prompt_result = await mcp.get_prompt(
38 | "WhatsApp Message Template",
39 | {"recipient": "John Doe", "context": "Following up on our meeting"}
40 | )
41 | print(f"Prompt Result: {prompt_result}")
42 | except Exception as e:
43 | print(f"Prompt test failed: {e}")
44 | print()
45 |
46 | # Test 4: Test tool calling (with mock data)
47 | print("🔧 Testing send_whatsapp_message tool:")
48 | try:
49 | # This will fail with API call but should show the structure
50 | result = await mcp.call_tool(
51 | "send_whatsapp_message",
52 | {
53 | "phone_number": "+1234567890",
54 | "message": "Test message from MCP server"
55 | }
56 | )
57 | print(f"Tool Result: {result}")
58 | except Exception as e:
59 | print(f"Tool test result (expected with test API key): {e}")
60 | print()
61 |
62 | print("✅ MCP Server structure test completed!")
63 |
64 | except Exception as e:
65 | print(f"❌ Error during testing: {e}")
66 | import traceback
67 | traceback.print_exc()
68 |
69 |
70 | def test_server_startup():
71 | """Test if the server can start properly"""
72 | print("🚀 Testing MCP Server Startup\n")
73 |
74 | try:
75 | # Test server initialization
76 | print("Server name:", mcp.name)
77 | print("Server initialized successfully!")
78 |
79 | # Run async tests
80 | asyncio.run(test_tools())
81 |
82 | except Exception as e:
83 | print(f"❌ Server startup failed: {e}")
84 | import traceback
85 | traceback.print_exc()
86 |
87 |
88 | if __name__ == "__main__":
89 | test_server_startup()
```
--------------------------------------------------------------------------------
/src/titan_mind/titan_mind_functions.py:
--------------------------------------------------------------------------------
```python
1 | from dataclasses import asdict
2 | from datetime import datetime, timedelta
3 | from typing import Any, Optional, Dict
4 |
5 | import requests
6 | from pydantic import BaseModel
7 |
8 | from titan_mind.utils.app_specific.networking.titan_mind import TitanMindAPINetworking, HTTPMethod
9 | from titan_mind.utils.general.date_time import get_date_time_to_utc_server_time_format_string
10 |
11 |
12 | # todo - it only returns 10 since page size is not given, rather than improving it add filter for search by phone number
13 | def get_conversations_from_the_last_day(
14 | phone_without_dialer_code: str = "None"
15 | ) -> Optional[Dict[str, Any]]:
16 | yesterday_datetime = datetime.now() - timedelta(days=1)
17 | payload = {
18 | "page": 1,
19 | "channel": "whatsapp",
20 | "last_message_at__gte": get_date_time_to_utc_server_time_format_string(yesterday_datetime)
21 | }
22 | if phone_without_dialer_code and phone_without_dialer_code.lower() not in ["none", "null"]:
23 | print(f"phone_without_dialer_code {phone_without_dialer_code}")
24 | payload["title__icontains"] = phone_without_dialer_code
25 | return asdict(
26 | TitanMindAPINetworking().make_request(
27 | endpoint=f"msg/conversations/",
28 | success_message="last 24 hours conversations fetched.",
29 | method=HTTPMethod.GET,
30 | payload=payload
31 | )
32 | )
33 |
34 |
35 | def get_the_conversation_messages(conversation_id: str) -> Optional[Dict[str, Any]]:
36 | return asdict(
37 | TitanMindAPINetworking().make_request(
38 | endpoint=f"msg/conversations/{conversation_id}/messages/",
39 | success_message="messages in a conversation fetched.",
40 | method=HTTPMethod.GET,
41 | payload={
42 | }
43 | )
44 | )
45 |
46 |
47 | def send_whatsapp_message_to_a_conversation(conversation_id: str, message: str):
48 | return asdict(
49 | TitanMindAPINetworking().make_request(
50 | endpoint=f"msg/conversations/{conversation_id}/messages/whatsapp/send-message/",
51 | success_message="whatsapp message sent request created.",
52 | method=HTTPMethod.POST,
53 | payload={
54 | "recipient_type": "individual",
55 | "type": "text",
56 | "text": {
57 | "body": message
58 | }
59 | }
60 | )
61 | )
62 |
63 |
64 | def register_msg_template_for_approval(
65 | template_name: str, language: str, category: str, message_content_components: list[dict[str, Any]]
66 | ):
67 | return asdict(
68 | TitanMindAPINetworking().make_request(
69 | endpoint=f"whatsapp/template/",
70 | payload={
71 | "name": template_name,
72 | "language": language,
73 | "category": category,
74 | "components": message_content_components
75 | },
76 | success_message="whatsapp template registered for approval.",
77 | method=HTTPMethod.POST,
78 | )
79 | )
80 |
81 |
82 | def get_the_templates(
83 | template_name: str = "None",
84 | page: int = 1,
85 | page_size: int = 10,
86 | ):
87 | payload = {
88 | "channel": "whatsapp",
89 | "page": page,
90 | "page_size": page_size
91 | }
92 | if template_name is not None and template_name.lower() not in ["none", "null"]:
93 | payload["name__icontains"] = template_name
94 |
95 | return asdict(
96 | TitanMindAPINetworking().make_request(
97 | endpoint=f"template/",
98 | payload=payload,
99 | success_message="templates fetched",
100 | method=HTTPMethod.GET,
101 | )
102 | )
103 |
104 |
105 | class Contact(BaseModel):
106 | country_code_alpha: str
107 | country_code: str
108 | phone_without_country_code: str
109 |
110 |
111 | def send_message_to_a_number_using_approved_template(
112 | template_id: int,
113 | contacts: list[Contact],
114 | ):
115 | return asdict(
116 | TitanMindAPINetworking().make_request(
117 | endpoint=f"whatsapp/message/send-template/",
118 | payload={
119 | "recipients": [contact.model_dump() for contact in contacts],
120 | "template": template_id,
121 | },
122 | success_message="message sent request created.",
123 | method=HTTPMethod.POST,
124 | )
125 | )
126 |
```
--------------------------------------------------------------------------------
/src/titan_mind/utils/app_specific/networking/titan_mind.py:
--------------------------------------------------------------------------------
```python
1 | from typing import Any, Dict, Optional, Callable, List, Union
2 | from dataclasses import dataclass, field
3 | from enum import Enum
4 | import requests
5 |
6 | from titan_mind.networking import get_titan_engage_headers, get_titan_engage_url, print_request_and_response
7 | from titan_mind.utils.app_specific.utils import to_run_mcp_in_server_mode_or_std_io
8 |
9 |
10 | # Enums for HTTP methods
11 | class HTTPMethod(Enum):
12 | GET = "GET"
13 | POST = "POST"
14 | PUT = "PUT"
15 | DELETE = "DELETE"
16 | PATCH = "PATCH"
17 |
18 |
19 | # Base response dataclasses
20 | @dataclass
21 | class BaseResponse:
22 | """Base response structure"""
23 | status: bool
24 | message: str
25 | result: Union[dict[str, Any], list[dict[str, Any]]] = field(default_factory=dict)
26 |
27 |
28 | class TitanMindAPINetworking:
29 | def __init__(self):
30 | self.base_headers = get_titan_engage_headers()
31 |
32 | def make_request(
33 | self,
34 | endpoint: str,
35 | payload: Dict[str, Any],
36 | success_message: str,
37 | method: HTTPMethod = HTTPMethod.POST,
38 | response_processor: Optional[Callable[[Dict[str, Any]], BaseResponse]] = None
39 | ) -> BaseResponse:
40 | """Internal method to handle all API requests"""
41 | response = None
42 | try:
43 | url = get_titan_engage_url(endpoint)
44 |
45 | # Use enum for method selection
46 | if method == HTTPMethod.POST:
47 | response = requests.post(url, headers=self.base_headers, json=payload)
48 | elif method == HTTPMethod.GET:
49 | response = requests.get(url, headers=self.base_headers, params=payload)
50 | elif method == HTTPMethod.PUT:
51 | response = requests.put(url, headers=self.base_headers, json=payload)
52 | elif method == HTTPMethod.DELETE:
53 | response = requests.delete(url, headers=self.base_headers, json=payload)
54 | elif method == HTTPMethod.PATCH:
55 | response = requests.patch(url, headers=self.base_headers, json=payload)
56 | else:
57 | return BaseResponse(
58 | status=False,
59 | message=f"Unsupported HTTP method: {method.value}",
60 | )
61 |
62 | if to_run_mcp_in_server_mode_or_std_io():
63 | # for some reason in stdio mode the printing json for a specific api is breaking the whole tool call, for so now disabling it for this mode
64 | print_request_and_response(response)
65 | response.raise_for_status()
66 |
67 | response_data = response.json()
68 |
69 | # Apply custom response processing if provided
70 | if response_processor:
71 | return response_processor(response_data)
72 |
73 | # Default response
74 | return BaseResponse(
75 | status=True,
76 | message=success_message,
77 | result=self.get_result_dict(response_data),
78 | )
79 |
80 | except requests.exceptions.HTTPError as e:
81 | error_json = {}
82 | if response is not None:
83 | try:
84 | error_json = response.json()
85 | except:
86 | error_json = {"error": "Could not parse error response", "status_code": response.status_code}
87 |
88 | return BaseResponse(
89 | status=False,
90 | message=f"HTTP Error: {str(e)}",
91 | result=error_json
92 | )
93 | except requests.exceptions.RequestException as e:
94 | return BaseResponse(
95 | status=False,
96 | message=f"Request Error: {str(e)}",
97 | result={"error_type": "RequestException"}
98 | )
99 | except Exception as e:
100 | error_json = {}
101 | if response is not None:
102 | try:
103 | error_json = response.json()
104 | except:
105 | error_json = {"error": "Could not parse response"}
106 |
107 | return BaseResponse(
108 | status=False,
109 | message=f"Unexpected Error: {str(e)}",
110 | result=error_json
111 | )
112 |
113 | def get_result_dict(self, response_data):
114 | return response_data.get("result", response_data)
115 |
```
--------------------------------------------------------------------------------
/src/titan_mind/server.py:
--------------------------------------------------------------------------------
```python
1 | import argparse
2 | import os
3 | from typing import Optional, Dict, Any, Callable, Annotated
4 |
5 | from fastmcp import FastMCP
6 |
7 | from titan_mind import titan_mind_functions as titan_mind_functions
8 | from titan_mind.titan_mind_functions import Contact
9 | from titan_mind.utils.app_specific.utils import to_run_mcp_in_server_mode_or_std_io, get_script_args
10 |
11 | # todo - atm below workflow is attached to every tool description, which is not recommended, but fastmcp and the claude desktop client are not able to share system instructions well. so for now appending the instructions to the tools context also
12 | # todo - give straightforward API or tool to determine if the receiver is in the free form window or not
13 | # todo - LLM are not following the usage instructions so need to update descriptions and check if mcp has other construct for it. the usage instruction given of --> checking if the the receiver have sent a message within the last 24 hours. and after sending the message check if the receiver recieved it or not.
14 | _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow = \
15 | """
16 | TITANMIND WHATSAPP WORKFLOW:
17 |
18 | FREE-FORM (24hr window):
19 | 1. Any content allowed
20 | 2. Only after user's have sent a message in the last 24 hours
21 |
22 | TEMPLATES (outside 24hr window):
23 | 1. Pre-approved structured content
24 | 2. Required for new conversations
25 |
26 | PROCESS:
27 | 1. Check receiver phone number free form messaging window status
28 | 2. A receiver is in the free-form messaging window if a conversation with their phone number already exists and also the receiver have sent a message within the last 24 hours.
29 | 2. Use free-form OR register template
30 | 3. Wait for template approval (if needed)
31 | 4. Send message
32 | 5. check the conversation, if the receiver received message successfully
33 | """
34 |
35 | _tool_return_object_description: str = \
36 | """
37 | Each tool Returns:
38 | a boolean if the tool was able to perform the api call successfully against the key "status"
39 | a string containing the message or error if the function ran successfully or not against the key "message"
40 | a dict containing the response according to the tool functionality or error details if the tool ran into an exception, against the key "result"
41 | """
42 |
43 | _mcp_name = "TitanMind Whatsapp MCP - Handles free-form messages (24hr window) and template workflows automatically"
44 |
45 | mcp = FastMCP(
46 | _mcp_name,
47 | instructions=_titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow + _tool_return_object_description,
48 | )
49 |
50 |
51 | @mcp.prompt()
52 | def whatsapp_and_server_workflow() -> str:
53 | """WhatsApp messaging workflow guide, and server tools return info"""
54 | return _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow + _tool_return_object_description
55 |
56 |
57 | @mcp.prompt()
58 | def send_whatsapp_message() -> str:
59 | """WhatsApp messaging workflow guide, and server tools return info"""
60 | return _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow + _tool_return_object_description
61 |
62 |
63 | @mcp.resource("resource://workflow")
64 | def whatsapp_and_server_workflow_resource() -> str:
65 | """WhatsApp messaging workflow guide, and server tools return info"""
66 | return _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow + _tool_return_object_description
67 |
68 | """
69 | Guidelines to follow:
70 | 1. Keep the tool names char count max to 51. Since some models do not support MCP, and rely on the function calling.
71 | So tool_name + mcp_name needs to be 64, making the mcp_name to have max 13 char count.
72 | """
73 | @mcp.tool()
74 | def get_conversations_from_the_last_day(
75 | phone_without_dialer_code: str = "None"
76 | ) -> Optional[Dict[str, Any]]:
77 | ("""
78 | get all the conversation where there have been the last message sent or received in the last 24 hours.
79 |
80 | Args:
81 | phone_without_dialer_code (str): to filter conversation with a phone number. default is "None" to get all conversations.
82 | """ + _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow)
83 |
84 | return titan_mind_functions.get_conversations_from_the_last_day(
85 | phone_without_dialer_code
86 | )
87 |
88 |
89 | @mcp.tool()
90 | def get_the_messages_of_a_conversation_(conversation_id: str) -> Optional[Dict[str, Any]]:
91 | ("""
92 | gets the messages in a conversation.
93 |
94 | Args:
95 | conversation_id (str): alphanumeric id of the whatsapp conversation, to which a message is required to be sent.
96 | message (str): the message to send.
97 | """ + _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow)
98 |
99 | return titan_mind_functions.get_the_conversation_messages(
100 | conversation_id
101 | )
102 |
103 |
104 | @mcp.tool()
105 | def send_whatsapp_message_to_a_conversation(conversation_id: str, message: str) -> Optional[Dict[str, Any]]:
106 | ("""
107 | sends a whatsapp message to a Titanmind's whatsapp conversation.
108 |
109 | Args:
110 | conversation_id (str): id of the whatsapp conversation.
111 | message (str): the message to send.
112 | """ + _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow)
113 |
114 | return titan_mind_functions.send_whatsapp_message_to_a_conversation(
115 | conversation_id, message
116 | )
117 |
118 |
119 | @mcp.tool()
120 | def register_msg_template_for_approval(
121 | template_name: str,
122 | message_content_components: list[dict[str, Any]],
123 | language: str = "en", category: str = "MARKETING",
124 | ) -> Optional[Dict[str, Any]]:
125 | """
126 | creates and registers a new whatsapp message template for approval.
127 | Args:
128 | template_name (str): name of the whatsapp message template, It only accepts a single word without no special characters except underscores
129 | language (str): language of the whatsapp message template (default is "en")
130 | category (str): category of the whatsapp message template (default is "MARKETING"), other possible values are "UTILITY", "AUTHENTICATION"
131 | message_content_components (dict): the message content that needs to be sent. It needs to be structured like the below example,
132 | components are required to have BODY component at least, like this: {"type": "BODY", "text": "lorem body text"}, BODY component is for the simple text.
133 | All other components are optional.
134 | HEADER component can have any of the below format, but only one format at a time can be used.: TEXT(the header component with TEXT needs to be like this
135 | {
136 | "type": "HEADER",
137 | "format": "TEXT",
138 | "text": "lorem header text"
139 | }
140 | ), VIDEO(the header component with VIDEO needs to be like this
141 | {
142 | "type":"HEADER",
143 | "format":"VIDEO",
144 | "example":{
145 | "header_handle":[
146 | "https://sample_video_url.jpg"
147 | ]
148 | }
149 | }
150 | )
151 | , IMAGE(the header component with IMAGE needs to be like this
152 | {
153 | "type":"HEADER",
154 | "format":"IMAGE",
155 | "example":{
156 | "header_handle":[
157 | "https://sample_image_url.jpg"
158 | ]
159 | }
160 | }),
161 | DOCUMENT (the header component with DOCUMENT needs to be like this
162 | {
163 | "type":"HEADER",
164 | "format":"DOCUMENT",
165 | "example":{
166 | "header_handle":[
167 | "https://sample_document_url"
168 | ]
169 | }
170 | }),
171 | message_content_components value with all other type of components is mentioned below.
172 | [
173 | {
174 | "type": "HEADER",
175 | "format": "TEXT",
176 | "text": "lorem header text"
177 | },
178 | {
179 | "type": "BODY",
180 | "text": "lorem body text"
181 | },
182 | {
183 | "type": "FOOTER",
184 | "text": "lorem footer text"
185 | },
186 | {
187 | "type": "BUTTONS",
188 | "buttons": [
189 | {
190 | "type": "QUICK_REPLY",
191 | "text": "lorem reply bt"
192 | },
193 | {
194 | "type": "URL",
195 | "text": "cta",
196 | "url": "https:sample.in"
197 | },
198 | {
199 | "type": "PHONE_NUMBER",
200 | "text": "call ",
201 | "phone_number": "IN328892398"
202 | }
203 | ]
204 | }
205 | ]
206 | Buttons need to follow order of first QUICK_REPLY, then URL, and then PHONE_NUMBER.
207 | """
208 |
209 | return titan_mind_functions.register_msg_template_for_approval(
210 | template_name, language, category, message_content_components
211 | )
212 |
213 |
214 | @mcp.tool()
215 | def get_the_templates(
216 | template_name: str = "None",
217 | page: int = 1,
218 | page_size: int = 10,
219 | ) -> Optional[Dict[str, Any]]:
220 | ("""
221 | gets all the created templates with the details like approved/pending status
222 |
223 | Args:
224 | template_name (str): name of the whatsapp message template, It only accepts a word without no special characters only underscores. Default is "None" to get all the templates
225 | page (int): page refers to the page in paginated api. default is 1
226 | page_size (int): page_size refers to the page_size in paginated api. default is 25
227 | """ + _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow)
228 | return titan_mind_functions.get_the_templates(
229 | template_name, page, page_size
230 | )
231 |
232 |
233 | @mcp.tool()
234 | def send_msg_to_multiple_num_using_approved_template(
235 | template_id: int, contacts: list[Contact],
236 | ) -> Optional[Dict[str, Any]]:
237 | ("""
238 | sends a message to a phone number using an approved whatsapp template.
239 |
240 | Args:
241 | template_id (str): id of the whatsapp message template, it is not the template name.
242 | contacts (Contact): a contact has three attributes: country_code_alpha(like "IN" for india), country_code(like "91") and phone_without_dialer_code
243 | """ + _titan_mind_product_whatsapp_channel_messaging_functionality_and_workflow)
244 | return titan_mind_functions.send_message_to_a_number_using_approved_template(
245 | template_id, contacts
246 | )
247 |
248 |
249 | def main():
250 | if to_run_mcp_in_server_mode_or_std_io():
251 | mcp.run(transport="streamable-http", host="0.0.0.0", port=get_script_args().port)
252 | else:
253 | mcp.run()
254 |
255 |
256 | if __name__ == "__main__":
257 | main()
258 |
```