"""TrustLayer API client for MCP server (read-only)."""
import httpx
from typing import Optional, Dict, Any
import logging
logger = logging.getLogger(__name__)
class TrustLayerClient:
"""Client for TrustLayer API (read-only operations)."""
def __init__(self, api_token: str, base_url: str = "https://api.trustlayer.io", api_version: str = "v1"):
"""Initialize TrustLayer client.
Args:
api_token: TrustLayer API token
base_url: Base URL for TrustLayer API (without version)
api_version: API version (e.g., "v1", "v2")
"""
self.api_token = api_token
self.base_url = base_url.rstrip("/")
self.api_version = api_version
self.headers = {
"Authorization": f"Bearer {api_token}",
"Content-Type": "application/json",
"Accept": "application/json",
}
def _make_request(
self, method: str, endpoint: str, params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Make HTTP request to TrustLayer API.
Args:
method: HTTP method (GET only for read-only)
endpoint: API endpoint
params: Query parameters
Returns:
Response data as dictionary
Raises:
httpx.HTTPError: If request fails
"""
if method != "GET":
raise ValueError("Only GET requests are allowed (read-only mode)")
# Add version prefix if endpoint doesn't start with /
if not endpoint.startswith("/"):
endpoint = f"/{endpoint}"
# Add API version prefix (e.g., /v1)
endpoint = f"/{self.api_version}{endpoint}"
url = f"{self.base_url}{endpoint}"
logger.debug(f"Making {method} request to {url}")
with httpx.Client(timeout=30.0) as client:
response = client.get(url, headers=self.headers, params=params)
response.raise_for_status()
return response.json()
def _get_all_pages(
self, endpoint: str, params: Optional[Dict[str, Any]] = None, max_pages: int = 100
) -> Dict[str, Any]:
"""Get all pages of paginated data.
Args:
endpoint: API endpoint
params: Query parameters (will be modified for pagination)
max_pages: Maximum number of pages to fetch (safety limit)
Returns:
Combined response with all data from all pages
"""
if params is None:
params = {}
# Start with first page
page_number = 1
all_data = []
total_items = None
# Use maximum page size (100) for efficiency when fetching all pages
page_size = 100
# Remove page parameters from params if they exist (we'll manage them)
params = {k: v for k, v in params.items() if not k.startswith("page[")}
while page_number <= max_pages:
# Set pagination parameters
page_params = params.copy()
page_params["page[number]"] = page_number
page_params["page[size]"] = page_size
response = self._make_request("GET", endpoint, page_params)
page_data = response.get("data", [])
all_data.extend(page_data)
# Get pagination metadata (TrustLayer API uses meta directly, not meta.pagination)
meta = response.get("meta", {})
# TrustLayer API format: {'count': X, 'pages': Y, 'totalCount': Z, 'totalPages': W}
total_pages = meta.get("totalPages", meta.get("pages", 1))
current_page = page_number
total_items = meta.get("totalCount", meta.get("count", len(all_data)))
logger.info(f"Fetched page {current_page}/{total_pages}, {len(page_data)} items (total so far: {len(all_data)})")
# Check if we've fetched all pages
if current_page >= total_pages or len(page_data) == 0:
break
page_number += 1
logger.info(f"Fetched all {len(all_data)} items from {page_number} pages")
# Return response in the same format as single page
return {
"data": all_data,
"meta": {
**meta,
"count": len(all_data),
"totalCount": len(all_data),
"pages": 1,
"totalPages": 1,
"fetched_all": True,
"pages_fetched": page_number,
}
}
# Parties endpoints
def get_parties(self, params: Optional[Dict[str, Any]] = None, fetch_all: bool = False) -> Dict[str, Any]:
"""Get list of parties.
Args:
params: Query parameters
fetch_all: If True, automatically fetch all pages
"""
if fetch_all:
return self._get_all_pages("/parties", params)
return self._make_request("GET", "/parties", params)
def get_party(self, party_id: str) -> Dict[str, Any]:
"""Get party by ID."""
return self._make_request("GET", f"/parties/{party_id}")
def get_party_contacts(self, party_id: str) -> Dict[str, Any]:
"""Get contacts for a party."""
return self._make_request("GET", f"/parties/{party_id}/contacts")
def get_party_compliance_profile(self, party_id: str) -> Dict[str, Any]:
"""Get compliance profile for a party."""
return self._make_request("GET", f"/parties/{party_id}/compliance-profile")
def get_party_compliance_certificate(self, party_id: str) -> Dict[str, Any]:
"""Get compliance certificate for a party."""
return self._make_request("GET", f"/parties/{party_id}/compliance-certificate")
def get_party_document_request(self, party_id: str) -> Dict[str, Any]:
"""Get document request for a party."""
return self._make_request("GET", f"/parties/{party_id}/document-request")
# Contacts endpoints
def get_contact(self, contact_id: str) -> Dict[str, Any]:
"""Get contact by ID."""
return self._make_request("GET", f"/contacts/{contact_id}")
# Documents endpoints
def get_documents(self, params: Optional[Dict[str, Any]] = None, fetch_all: bool = False) -> Dict[str, Any]:
"""Get list of documents.
Args:
params: Query parameters
fetch_all: If True, automatically fetch all pages
"""
if fetch_all:
return self._get_all_pages("/documents", params)
return self._make_request("GET", "/documents", params)
def get_document(self, document_id: str) -> Dict[str, Any]:
"""Get document by ID."""
return self._make_request("GET", f"/documents/{document_id}")
# Projects endpoints
def get_projects(self, params: Optional[Dict[str, Any]] = None, fetch_all: bool = False) -> Dict[str, Any]:
"""Get list of projects.
Args:
params: Query parameters
fetch_all: If True, automatically fetch all pages
"""
if fetch_all:
return self._get_all_pages("/projects", params)
return self._make_request("GET", "/projects", params)
def get_project(self, project_id: str) -> Dict[str, Any]:
"""Get project by ID."""
return self._make_request("GET", f"/projects/{project_id}")
# Tags endpoints
def get_tags(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get list of tags."""
return self._make_request("GET", "/tags", params)
# Party Types endpoints
def get_party_types(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get list of party types."""
return self._make_request("GET", "/party-types", params)
# Compliance Profiles endpoints
def get_compliance_profiles(
self, params: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Get list of compliance profiles."""
return self._make_request("GET", "/compliance-profiles", params)
def get_compliance_profile(self, profile_id: str) -> Dict[str, Any]:
"""Get compliance profile by ID."""
return self._make_request("GET", f"/compliance-profiles/{profile_id}")
# Custom Fields endpoints
def get_custom_fields(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get list of custom fields."""
return self._make_request("GET", "/custom-fields", params)
# Reports endpoints
def get_reports(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get list of reports."""
return self._make_request("GET", "/reports", params)
def get_report(self, report_id: str) -> Dict[str, Any]:
"""Get report by ID."""
return self._make_request("GET", f"/reports/{report_id}")
# Webhooks endpoints
def get_webhooks(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get list of webhooks."""
return self._make_request("GET", "/webhooks", params)
# Branding endpoints
def get_branding(self, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Get list of branding."""
return self._make_request("GET", "/branding", params)
def get_branding_item(self, branding_id: str) -> Dict[str, Any]:
"""Get branding by ID."""
return self._make_request("GET", f"/branding/{branding_id}")
