aa397c585e
MCP server wrapping the TPP Wholesale domain registrar HTTP API (v2.7.4). Provides 13 MCP tools for domain registration, DNS management, renewals, transfers, contact management, and account queries. Features: - Async TPP API client with automatic session management - FastMCP server with Streamable HTTP transport - Docker deployment with docker-compose - Full .com.au eligibility support - REST health check and balance endpoints Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
470 lines
15 KiB
Python
470 lines
15 KiB
Python
"""TPP Wholesale MCP Server.
|
|
|
|
Exposes TPP Wholesale domain registrar operations as MCP tools.
|
|
Supports both Streamable HTTP and SSE transports.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
from mcp.server.fastmcp.utilities.types import TransportSecuritySettings
|
|
|
|
from tpp_client import TPPClient, TPPError
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger("tppwholesale-mcp")
|
|
|
|
# --- Configuration -----------------------------------------------------------
|
|
|
|
ACCOUNT_NO = os.environ.get("TPP_ACCOUNT_NO", "")
|
|
USER_ID = os.environ.get("TPP_USER_ID", "")
|
|
PASSWORD = os.environ.get("TPP_PASSWORD", "")
|
|
ALLOWED_HOSTS = os.environ.get(
|
|
"TPP_ALLOWED_HOSTS",
|
|
"localhost:8089,127.0.0.1:8089",
|
|
).split(",")
|
|
|
|
if not all([ACCOUNT_NO, USER_ID, PASSWORD]):
|
|
logger.warning(
|
|
"TPP_ACCOUNT_NO, TPP_USER_ID, and TPP_PASSWORD must be set. "
|
|
"The server will start but all tools will fail."
|
|
)
|
|
|
|
# --- MCP Server --------------------------------------------------------------
|
|
|
|
mcp = FastMCP(
|
|
"TPP Wholesale Domain Registrar",
|
|
transport_security=TransportSecuritySettings(
|
|
allowed_hosts=[h.strip() for h in ALLOWED_HOSTS],
|
|
),
|
|
)
|
|
|
|
_client: TPPClient | None = None
|
|
|
|
|
|
def _get_client() -> TPPClient:
|
|
global _client
|
|
if _client is None:
|
|
_client = TPPClient(ACCOUNT_NO, USER_ID, PASSWORD)
|
|
return _client
|
|
|
|
|
|
def _error_response(e: Exception) -> str:
|
|
if isinstance(e, TPPError):
|
|
return json.dumps({"error": True, "code": e.code, "message": e.message})
|
|
return json.dumps({"error": True, "message": str(e)})
|
|
|
|
|
|
# =============================================================================
|
|
# MCP Tools
|
|
# =============================================================================
|
|
|
|
|
|
@mcp.tool()
|
|
async def check_domain_availability(domains: str) -> str:
|
|
"""Check if one or more domains are available for registration.
|
|
|
|
Args:
|
|
domains: Comma-separated list of fully qualified domain names
|
|
(e.g. "example.com,example.com.au,example.net")
|
|
"""
|
|
try:
|
|
domain_list = [d.strip() for d in domains.split(",") if d.strip()]
|
|
client = _get_client()
|
|
results = await client.check_availability(domain_list)
|
|
return json.dumps(results, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def register_domain(
|
|
domain: str,
|
|
period: int,
|
|
owner_contact_id: str,
|
|
admin_contact_id: str = "",
|
|
tech_contact_id: str = "",
|
|
billing_contact_id: str = "",
|
|
nameservers: str = "",
|
|
registrant_name: str = "",
|
|
registrant_id: str = "",
|
|
registrant_id_type: str = "",
|
|
eligibility_type: str = "",
|
|
eligibility_reason: str = "",
|
|
) -> str:
|
|
"""Register a new domain name.
|
|
|
|
For .com.au domains, the registrant_name, registrant_id,
|
|
registrant_id_type, eligibility_type, and eligibility_reason
|
|
fields are mandatory.
|
|
|
|
Args:
|
|
domain: Fully qualified domain name (e.g. "example.com.au")
|
|
period: Registration period in years
|
|
owner_contact_id: Contact ID for domain owner (from create_contact)
|
|
admin_contact_id: Contact ID for admin (defaults to owner)
|
|
tech_contact_id: Contact ID for tech (defaults to owner)
|
|
billing_contact_id: Contact ID for billing (defaults to owner)
|
|
nameservers: Comma-separated nameservers (e.g. "ns1.cloudflare.com,ns2.cloudflare.com")
|
|
registrant_name: Legal entity name (required for .au)
|
|
registrant_id: ABN/ACN of the legal entity (required for .au)
|
|
registrant_id_type: ID type code — ABN, ACN, TM etc (required for .au)
|
|
eligibility_type: Eligibility type code (required for .au)
|
|
eligibility_reason: Eligibility reason code (required for .au)
|
|
"""
|
|
try:
|
|
hosts = [h.strip() for h in nameservers.split(",") if h.strip()] if nameservers else None
|
|
client = _get_client()
|
|
order_id = await client.register_domain(
|
|
domain=domain,
|
|
period=period,
|
|
owner_contact_id=owner_contact_id,
|
|
admin_contact_id=admin_contact_id or None,
|
|
tech_contact_id=tech_contact_id or None,
|
|
billing_contact_id=billing_contact_id or None,
|
|
hosts=hosts,
|
|
registrant_name=registrant_name or None,
|
|
registrant_id=registrant_id or None,
|
|
registrant_id_type=registrant_id_type or None,
|
|
eligibility_type=eligibility_type or None,
|
|
eligibility_reason=eligibility_reason or None,
|
|
)
|
|
return json.dumps({
|
|
"success": True,
|
|
"order_id": order_id,
|
|
"domain": domain,
|
|
"message": f"Domain {domain} registration submitted. Use check_order_status to track progress.",
|
|
}, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def create_contact(
|
|
first_name: str,
|
|
last_name: str,
|
|
email: str,
|
|
phone: str,
|
|
address1: str,
|
|
city: str,
|
|
region: str,
|
|
postal_code: str,
|
|
country_code: str,
|
|
organisation: str = "",
|
|
address2: str = "",
|
|
) -> str:
|
|
"""Create a contact object for use in domain registrations.
|
|
|
|
Returns a contact ID that can be used as owner/admin/tech/billing
|
|
contact when registering domains.
|
|
|
|
Args:
|
|
first_name: Contact first name
|
|
last_name: Contact last name
|
|
email: Contact email address
|
|
phone: Phone number (international format, e.g. "+61.7XXXXXXXX")
|
|
address1: Street address line 1
|
|
city: City
|
|
region: State or region (e.g. "QLD")
|
|
postal_code: Postal/ZIP code
|
|
country_code: 2-letter country code (e.g. "AU")
|
|
organisation: Organisation name (optional)
|
|
address2: Street address line 2 (optional)
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
contact_id = await client.create_contact(
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
email=email,
|
|
phone=phone,
|
|
address1=address1,
|
|
city=city,
|
|
region=region,
|
|
postal_code=postal_code,
|
|
country_code=country_code,
|
|
organisation=organisation,
|
|
address2=address2,
|
|
)
|
|
return json.dumps({
|
|
"success": True,
|
|
"contact_id": contact_id,
|
|
"message": "Contact created. Use this ID for domain registrations.",
|
|
}, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def renew_domain(domain: str, period: int = 1) -> str:
|
|
"""Renew an existing domain registration.
|
|
|
|
Args:
|
|
domain: Fully qualified domain name to renew
|
|
period: Renewal period in years (default: 1)
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
order_id = await client.renew_domain(domain, period)
|
|
return json.dumps({
|
|
"success": True,
|
|
"order_id": order_id,
|
|
"domain": domain,
|
|
"period": period,
|
|
"message": f"Domain {domain} renewal submitted for {period} year(s).",
|
|
}, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def get_domain_details(domain: str) -> str:
|
|
"""Get full details for a domain including contacts, nameservers, and status.
|
|
|
|
Args:
|
|
domain: Fully qualified domain name
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
details = await client.get_domain_details(domain)
|
|
return json.dumps(details, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def list_domains(
|
|
search: str = "",
|
|
tld: str = "",
|
|
active_only: bool = True,
|
|
expires_before: str = "",
|
|
expires_after: str = "",
|
|
) -> str:
|
|
"""List domains in the TPP Wholesale account.
|
|
|
|
Args:
|
|
search: Filter by domain name substring (optional)
|
|
tld: Filter by TLD, e.g. ".com.au" (optional)
|
|
active_only: Only show active domains (default: true)
|
|
expires_before: Filter by expiry date, format YYYY-MM-DD (optional)
|
|
expires_after: Filter by expiry date, format YYYY-MM-DD (optional)
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
domains = await client.list_domains(
|
|
search=search or None,
|
|
tld=tld or None,
|
|
active_only=active_only,
|
|
expires_before=expires_before or None,
|
|
expires_after=expires_after or None,
|
|
)
|
|
return json.dumps({
|
|
"count": len(domains),
|
|
"domains": domains,
|
|
}, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def update_nameservers(
|
|
domain: str,
|
|
add: str = "",
|
|
remove: str = "",
|
|
) -> str:
|
|
"""Update nameservers on a domain.
|
|
|
|
Args:
|
|
domain: Fully qualified domain name
|
|
add: Comma-separated nameservers to add (e.g. "ns1.cloudflare.com,ns2.cloudflare.com")
|
|
remove: Comma-separated nameservers to remove
|
|
"""
|
|
try:
|
|
add_hosts = [h.strip() for h in add.split(",") if h.strip()] if add else None
|
|
remove_hosts = [h.strip() for h in remove.split(",") if h.strip()] if remove else None
|
|
|
|
if not add_hosts and not remove_hosts:
|
|
return json.dumps({"error": True, "message": "Provide at least one nameserver to add or remove."})
|
|
|
|
client = _get_client()
|
|
result = await client.update_hosts(domain, add_hosts, remove_hosts)
|
|
return json.dumps({
|
|
"success": True,
|
|
"domain": domain,
|
|
"added": add_hosts or [],
|
|
"removed": remove_hosts or [],
|
|
"order_id": result,
|
|
}, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def set_domain_lock(domain: str, lock: bool) -> str:
|
|
"""Lock or unlock a domain to prevent unauthorized transfers.
|
|
|
|
Args:
|
|
domain: Fully qualified domain name
|
|
lock: True to lock, False to unlock
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
result = await client.set_domain_lock(domain, lock)
|
|
return json.dumps({
|
|
"success": True,
|
|
"domain": domain,
|
|
"locked": lock,
|
|
"order_id": result,
|
|
}, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def check_order_status(order_id: str) -> str:
|
|
"""Check the status of a domain order (registration, renewal, transfer).
|
|
|
|
Args:
|
|
order_id: The order ID returned from register/renew/transfer operations
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
status = await client.get_order_status(order_id)
|
|
return json.dumps(status, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def get_account_balance() -> str:
|
|
"""Get the current prepaid account balance in AUD."""
|
|
try:
|
|
client = _get_client()
|
|
balance = await client.get_account_balance()
|
|
return json.dumps({
|
|
"balance_aud": balance,
|
|
"message": f"Account balance: ${balance:.2f} AUD",
|
|
}, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def get_renewal_info(domain: str) -> str:
|
|
"""Check renewal status and expiry date for a domain.
|
|
|
|
Args:
|
|
domain: Fully qualified domain name
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
info = await client.get_renewal_info(domain)
|
|
return json.dumps(info, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def check_transfer(domain: str, password: str = "") -> str:
|
|
"""Check if a domain can be transferred from another registrar.
|
|
|
|
Args:
|
|
domain: Fully qualified domain name
|
|
password: Domain auth/EPP code (required for some TLDs)
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
result = await client.check_transfer(domain, password or None)
|
|
return json.dumps(result, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
@mcp.tool()
|
|
async def transfer_domain(
|
|
domain: str,
|
|
period: int,
|
|
domain_password: str,
|
|
owner_contact_id: str,
|
|
admin_contact_id: str = "",
|
|
tech_contact_id: str = "",
|
|
billing_contact_id: str = "",
|
|
) -> str:
|
|
"""Request a domain transfer from another registrar.
|
|
|
|
Args:
|
|
domain: Fully qualified domain name to transfer
|
|
period: Renewal period in years (included with transfer)
|
|
domain_password: EPP/auth code from current registrar
|
|
owner_contact_id: Contact ID for domain owner
|
|
admin_contact_id: Contact ID for admin (defaults to owner)
|
|
tech_contact_id: Contact ID for tech (defaults to owner)
|
|
billing_contact_id: Contact ID for billing (defaults to owner)
|
|
"""
|
|
try:
|
|
client = _get_client()
|
|
order_id = await client.transfer_domain(
|
|
domain=domain,
|
|
period=period,
|
|
domain_password=domain_password,
|
|
owner_contact_id=owner_contact_id,
|
|
admin_contact_id=admin_contact_id or None,
|
|
tech_contact_id=tech_contact_id or None,
|
|
billing_contact_id=billing_contact_id or None,
|
|
)
|
|
return json.dumps({
|
|
"success": True,
|
|
"order_id": order_id,
|
|
"domain": domain,
|
|
"message": f"Transfer request submitted for {domain}. Transfers can take up to 9 days.",
|
|
}, indent=2)
|
|
except Exception as e:
|
|
return _error_response(e)
|
|
|
|
|
|
# =============================================================================
|
|
# REST API endpoints (non-MCP, for direct HTTP access)
|
|
# =============================================================================
|
|
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
|
|
app = mcp.get_app()
|
|
|
|
|
|
@app.get("/api/health")
|
|
async def health(request: Request) -> JSONResponse:
|
|
"""Health check endpoint."""
|
|
configured = all([ACCOUNT_NO, USER_ID, PASSWORD])
|
|
return JSONResponse({
|
|
"status": "ok" if configured else "unconfigured",
|
|
"service": "tppwholesale-mcp",
|
|
"configured": configured,
|
|
})
|
|
|
|
|
|
@app.get("/api/balance")
|
|
async def api_balance(request: Request) -> JSONResponse:
|
|
"""Quick balance check via REST."""
|
|
api_key = request.headers.get("x-api-key", "")
|
|
expected_key = os.environ.get("TPP_API_KEY", "")
|
|
if expected_key and api_key != expected_key:
|
|
return JSONResponse({"error": "Unauthorized"}, status_code=401)
|
|
|
|
try:
|
|
client = _get_client()
|
|
balance = await client.get_account_balance()
|
|
return JSONResponse({"balance_aud": balance})
|
|
except Exception as e:
|
|
return JSONResponse({"error": str(e)}, status_code=500)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
port = int(os.environ.get("TPP_PORT", "8089"))
|
|
uvicorn.run(app, host="0.0.0.0", port=port)
|