"""MCP tools for TrustLayer API (read-only operations)."""
from mcp.types import Tool, TextContent
from typing import List, Dict, Any, Optional
import json
from src.trustlayer_client import TrustLayerClient
from src.config import settings, api_token_context
def _build_query_params(arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Build query parameters according to TrustLayer API format.
Converts user-friendly parameters to API format:
- page_number -> page[number]
- page_size -> page[size]
- filter dict -> filter[key]=value format (deepObject style)
- sort, include remain as strings with comma-separated values
Args:
arguments: Tool arguments dictionary
Returns:
Dictionary with properly formatted query parameters
"""
params: Dict[str, Any] = {}
# Pagination: convert to page[number] and page[size] format
if "page_number" in arguments:
params["page[number]"] = arguments["page_number"]
if "page_size" in arguments:
params["page[size]"] = arguments["page_size"]
# Sort: comma-separated string (e.g. "name,-createdAt")
if "sort" in arguments:
params["sort"] = arguments["sort"]
# Include: comma-separated string (e.g. "parties,projects")
if "include" in arguments:
params["include"] = arguments["include"]
# Filter: deepObject style with square brackets
# httpx will automatically encode filter[title]=foo format
if "filter" in arguments and isinstance(arguments["filter"], dict):
filter_dict = arguments["filter"]
for key, value in filter_dict.items():
params[f"filter[{key}]"] = value
return params
def get_client() -> TrustLayerClient:
"""Get TrustLayer client instance.
Token priority:
1. From HTTP request header (Authorization) - for HTTP mode
2. From environment/config - for stdio mode fallback
"""
# Try to get token from request context (HTTP mode)
token = api_token_context.get()
# Fallback to config token (stdio mode or if no header provided)
if not token:
token = settings.trustlayer_api_token
if not token:
raise ValueError(
"API token is required. "
"For HTTP mode: provide Authorization header with Bearer token. "
"For stdio mode: set TRUSTLAYER_API_TOKEN environment variable."
)
# Extract token from "Bearer <token>" format if needed
if token.startswith("Bearer "):
token = token[7:]
return TrustLayerClient(
api_token=token,
base_url=settings.trustlayer_api_base_url,
api_version=settings.trustlayer_api_version,
)
def get_tools() -> List[Tool]:
"""Get list of available MCP tools."""
return [
Tool(
name="get_party",
description="Get a party by ID from TrustLayer",
inputSchema={
"type": "object",
"properties": {
"party_id": {
"type": "string",
"description": "The ID of the party to retrieve",
}
},
"required": ["party_id"],
},
),
Tool(
name="get_document",
description="Get a document by ID from TrustLayer",
inputSchema={
"type": "object",
"properties": {
"document_id": {
"type": "string",
"description": "The ID of the document to retrieve",
}
},
"required": ["document_id"],
},
),
Tool(
name="get_project",
description="Get a project by ID from TrustLayer",
inputSchema={
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "The ID of the project to retrieve",
}
},
"required": ["project_id"],
},
),
Tool(
name="get_contact",
description="Get a contact by ID from TrustLayer",
inputSchema={
"type": "object",
"properties": {
"contact_id": {
"type": "string",
"description": "The ID of the contact to retrieve",
}
},
"required": ["contact_id"],
},
),
Tool(
name="get_party_contacts",
description="Get all contacts for a party",
inputSchema={
"type": "object",
"properties": {
"party_id": {
"type": "string",
"description": "The ID of the party",
}
},
"required": ["party_id"],
},
),
Tool(
name="get_party_compliance_profile",
description="Get compliance profile for a party",
inputSchema={
"type": "object",
"properties": {
"party_id": {
"type": "string",
"description": "The ID of the party",
}
},
"required": ["party_id"],
},
),
Tool(
name="get_party_compliance_certificate",
description="Get compliance certificate for a party",
inputSchema={
"type": "object",
"properties": {
"party_id": {
"type": "string",
"description": "The ID of the party",
}
},
"required": ["party_id"],
},
),
Tool(
name="get_party_document_request",
description="Get document request for a party",
inputSchema={
"type": "object",
"properties": {
"party_id": {
"type": "string",
"description": "The ID of the party",
}
},
"required": ["party_id"],
},
),
Tool(
name="get_compliance_profile",
description="Get a compliance profile by ID",
inputSchema={
"type": "object",
"properties": {
"profile_id": {
"type": "string",
"description": "The ID of the compliance profile",
}
},
"required": ["profile_id"],
},
),
Tool(
name="get_report",
description="Get a report by ID",
inputSchema={
"type": "object",
"properties": {
"report_id": {
"type": "string",
"description": "The ID of the report",
}
},
"required": ["report_id"],
},
),
Tool(
name="list_parties",
description="List parties with smart search optimization. SEARCH STRATEGY: 1) With filters: starts with targeted search (20 results), expands to all pages if results found 2) If no results with filters, tries simplified filters 3) If sorting specified, fetches all pages for complete sorted dataset 4) Without filters/sort/pagination: fetches ALL parties (for 'list all' requests) 5) With explicit pagination: returns requested page only. Use filters for targeted searches, then fetch details with get_party if needed.\n\nAvailable filters: project, name, type, status, createdAt, updatedAt, customField, externalId\nAvailable sort keys: name, createdAt, updatedAt, status\nAvailable include options: projects, documents, tags, complianceProfile, customFields\n\nFilter notes:\n- type filter accepts singular or plural (e.g. 'Vendor' or 'Vendors')\n- status filter values: 'non_compliant', 'compliant'\n- externalId format: '<provider name>/<external id>' (e.g. 'procore/1234')\n- customField format: '<custom field id>;<field value>' (exact match, case-sensitive)",
inputSchema={
"type": "object",
"properties": {
"page_number": {
"type": "integer",
"description": "Page number (starting from 1). Note: If filter or sort is used, all pages are automatically fetched regardless of this parameter.",
"minimum": 1,
"default": 1,
},
"page_size": {
"type": "integer",
"description": "Number of items per page (max 100). Note: If filter or sort is used, all pages are automatically fetched regardless of this parameter.",
"minimum": 1,
"maximum": 100,
"default": 20,
},
"sort": {
"type": "string",
"description": "Comma-separated sort keys (prepend '-' for descending), e.g. 'name,-createdAt'. Available: name, createdAt, updatedAt, status. When used, ALL matching pages are automatically fetched for complete analysis.",
},
"filter": {
"type": "object",
"description": "Filter parameters. Available filters: project, name, type, status, createdAt, updatedAt, customField, externalId. When used, ALL matching pages are automatically fetched for complete analysis.",
},
"include": {
"type": "string",
"description": "Comma-separated relationships to include. Available: projects, documents, tags, complianceProfile, customFields",
},
},
},
),
Tool(
name="list_documents",
description="List documents with smart search optimization. SEARCH STRATEGY: 1) With filters: starts with targeted search (20 results), expands to all pages if results found 2) If no results with filters, tries simplified filters 3) If sorting specified, fetches all pages for complete sorted dataset 4) Without filters/sort/pagination: fetches ALL documents (for 'list all' requests) 5) With explicit pagination: returns requested page only. Use filters for targeted searches, then fetch details with get_document if needed.\n\nAvailable filters: name, party, project, type, archived, reviewed, createdAt, updatedAt, reviewedAt, archivedAt, expirationDate\nAvailable sort keys: name, createdAt, updatedAt, archivedAt, reviewedAt, expirationDate\nAvailable include options: party, projects, reviewedBy, archivedBy",
inputSchema={
"type": "object",
"properties": {
"page_number": {
"type": "integer",
"description": "Page number (starting from 1). Note: If filter or sort is used, all pages are automatically fetched regardless of this parameter.",
"minimum": 1,
"default": 1,
},
"page_size": {
"type": "integer",
"description": "Number of items per page (max 100). Note: If filter or sort is used, all pages are automatically fetched regardless of this parameter.",
"minimum": 1,
"maximum": 100,
"default": 20,
},
"sort": {
"type": "string",
"description": "Comma-separated sort keys (prepend '-' for descending), e.g. 'name,-createdAt'. Available: name, createdAt, updatedAt, archivedAt, reviewedAt, expirationDate. When used, ALL matching pages are automatically fetched for complete analysis.",
},
"filter": {
"type": "object",
"description": "Filter parameters. Available filters: name, party, project, type, archived, reviewed, createdAt, updatedAt, reviewedAt, archivedAt, expirationDate. When used, ALL matching pages are automatically fetched for complete analysis.",
},
"include": {
"type": "string",
"description": "Comma-separated relationships to include. Available: party, projects, reviewedBy, archivedBy",
},
},
},
),
Tool(
name="list_projects",
description="List projects with smart search optimization. SEARCH STRATEGY: 1) With filters: starts with targeted search (20 results), expands to all pages if results found 2) If no results with filters, tries simplified filters 3) If sorting specified, fetches all pages for complete sorted dataset 4) Without filters/sort/pagination: fetches ALL projects (for 'list all' requests) 5) With explicit pagination: returns requested page only. Use filters for targeted searches, then fetch details with get_project if needed.\n\nAvailable filters: name, status, active, createdAt, updatedAt, startDate, endDate, customField, externalId\nAvailable sort keys: name, createdAt, updatedAt, startDate, endDate\nAvailable include options: parties, customFields\n\nFilter notes:\n- status filter values: 'compliant', 'non_compliant', and others\n- externalId format: '<provider name>/<external id>' (e.g. 'procore/1234')\n- customField format: '<custom field id>/<field value>' (exact match, eq operator only)",
inputSchema={
"type": "object",
"properties": {
"page_number": {
"type": "integer",
"description": "Page number (starting from 1). Note: If filter or sort is used, all pages are automatically fetched regardless of this parameter.",
"minimum": 1,
"default": 1,
},
"page_size": {
"type": "integer",
"description": "Number of items per page (max 100). Note: If filter or sort is used, all pages are automatically fetched regardless of this parameter.",
"minimum": 1,
"maximum": 100,
"default": 20,
},
"sort": {
"type": "string",
"description": "Comma-separated sort keys (prepend '-' for descending), e.g. 'name,-createdAt'. Available: name, createdAt, updatedAt, startDate, endDate. When used, ALL matching pages are automatically fetched for complete analysis.",
},
"filter": {
"type": "object",
"description": "Filter parameters. Available filters: name, status, active, createdAt, updatedAt, startDate, endDate, customField, externalId. When used, ALL matching pages are automatically fetched for complete analysis.",
},
"include": {
"type": "string",
"description": "Comma-separated relationships to include. Available: parties, customFields",
},
},
},
),
Tool(
name="list_tags",
description="List all tags",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="list_compliance_profiles",
description="List all compliance profiles with optional filters, sorting, and pagination.\n\nAvailable filters: name, moduleId, subjectId, createdAt, updatedAt\nAvailable sort keys: name, createdAt, updatedAt\n\nFilter notes:\n- moduleId and subjectId filters only perform exact match (case-sensitive); they do not support other operators",
inputSchema={
"type": "object",
"properties": {
"page_number": {
"type": "integer",
"description": "Page number (starting from 1)",
"minimum": 1,
"default": 1,
},
"page_size": {
"type": "integer",
"description": "Number of items per page (max 100)",
"minimum": 1,
"maximum": 100,
"default": 20,
},
"sort": {
"type": "string",
"description": "Comma-separated sort keys (prepend '-' for descending). Available: name, createdAt, updatedAt",
},
"filter": {
"type": "object",
"description": "Filter parameters. Available filters: name, moduleId, subjectId, createdAt, updatedAt",
},
},
},
),
Tool(
name="list_party_types",
description="List all party types",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="list_custom_fields",
description="List all custom fields",
inputSchema={"type": "object", "properties": {}},
),
Tool(
name="list_reports",
description="List all public reports with optional filters and sorting.\n\nAvailable filters: type (either 'projects' or 'parties')\nAvailable sort keys: name, createdAt",
inputSchema={
"type": "object",
"properties": {
"sort": {
"type": "string",
"description": "Comma-separated sort keys (prepend '-' for descending). Available: name, createdAt",
},
"filter": {
"type": "object",
"description": "Filter parameters. Available filters: type (either 'projects' or 'parties')",
},
},
},
),
Tool(
name="list_webhooks",
description="List all webhooks with optional filters.\n\nAvailable filters: namespace",
inputSchema={
"type": "object",
"properties": {
"filter": {
"type": "object",
"description": "Filter parameters. Available filters: namespace",
},
},
},
),
Tool(
name="get_branding",
description="Get a branding configuration by ID",
inputSchema={
"type": "object",
"properties": {
"branding_id": {
"type": "string",
"description": "The ID of the branding configuration to retrieve",
}
},
"required": ["branding_id"],
},
),
Tool(
name="list_branding",
description="List all branding configurations with optional filters, sorting, and pagination.\n\nAvailable filters: name, createdAt, updatedAt\nAvailable sort keys: name, createdAt, updatedAt",
inputSchema={
"type": "object",
"properties": {
"page_number": {
"type": "integer",
"description": "Page number (starting from 1)",
"minimum": 1,
"default": 1,
},
"page_size": {
"type": "integer",
"description": "Number of items per page (max 100)",
"minimum": 1,
"maximum": 100,
"default": 20,
},
"sort": {
"type": "string",
"description": "Comma-separated sort keys (prepend '-' for descending). Available: name, createdAt, updatedAt",
},
"filter": {
"type": "object",
"description": "Filter parameters. Available filters: name, createdAt, updatedAt",
},
},
},
),
]
def _smart_list_search(
client: TrustLayerClient,
list_method,
arguments: Dict[str, Any],
max_initial_results: int = 20
) -> Dict[str, Any]:
"""
Smart search algorithm that optimizes API usage while ensuring NO DATA LOSS:
1. First try targeted search with filters (single page check)
2. If results found, ALWAYS fetch ALL pages with filters (complete dataset)
3. If no results, try with simplified filters
4. If sorting specified, fetch all pages (complete sorted dataset)
5. Without filters/sort/pagination: fetch ALL (for "list all" requests)
6. Only with explicit pagination: return single page
CRITICAL: This function ensures complete data retrieval - no records are lost.
Args:
client: TrustLayer client instance
list_method: Method to call (e.g., client.get_parties)
arguments: Tool arguments
max_initial_results: Maximum results for initial targeted search check
Returns:
API response with data (complete dataset, no pagination loss)
"""
params = _build_query_params(arguments)
filters = arguments.get("filter", {})
has_filters = bool(filters)
has_sort = bool(arguments.get("sort"))
explicit_pagination = "page_number" in arguments or "page_size" in arguments
# Strategy 1: If we have filters, start with targeted search (single page, max size)
if has_filters and not explicit_pagination:
# First attempt: targeted search with all filters, max page size
# Remove any existing pagination params to ensure we control them
targeted_params = {k: v for k, v in params.items() if not k.startswith("page[")}
targeted_params["page[number]"] = 1
targeted_params["page[size]"] = max_initial_results
result = list_method(targeted_params, fetch_all=False)
result_data = result.get("data", [])
# If we got results, fetch ALL pages with this filter
# This ensures complete filtered dataset - NO DATA LOSS
# Handles cases like "find all parties with status=compliant" correctly
if result_data:
# Always fetch all pages when filters are used and results found
# This guarantees we return ALL matching records, not just first page
return list_method(params, fetch_all=True)
# Strategy 2: If no results with all filters, try with simplified filters
# Remove less specific filters, keep most important ones
if len(filters) > 1:
# Priority order: name > type > status > party > project > others
priority_filters = ["name", "type", "status", "party", "project"]
simplified_filters = {}
for key in priority_filters:
if key in filters:
simplified_filters[key] = filters[key]
break # Try with just one priority filter first
if simplified_filters:
# Build new params without old filters
simplified_params = {}
# Copy non-filter params (sort, include, etc.)
for k, v in params.items():
if not k.startswith("filter["):
simplified_params[k] = v
# Add simplified filter
for key, value in simplified_filters.items():
simplified_params[f"filter[{key}]"] = value
simplified_params["page[number]"] = 1
simplified_params["page[size]"] = max_initial_results
result = list_method(simplified_params, fetch_all=False)
if result.get("data"):
# Found something with simplified filter, fetch all pages
return list_method(simplified_params, fetch_all=True)
# Strategy 3: No filters or explicit pagination requested
# CRITICAL: Ensure NO DATA LOSS - fetch all pages unless explicitly paginated
if has_sort:
# Sorting means user wants complete sorted dataset - fetch ALL pages
return list_method(params, fetch_all=True)
elif explicit_pagination:
# User explicitly requested specific page - respect pagination
return list_method(params, fetch_all=False)
else:
# No filters, no sort, no pagination: user wants ALL data
# This handles cases like "list all parties" or "show me all documents"
# CRITICAL: Fetch ALL pages to ensure complete dataset - NO DATA LOSS
return list_method(params, fetch_all=True)
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
"""Call a tool by name with arguments.
Uses smart search algorithm for list operations to optimize API usage:
- Targeted searches with filters (single page first)
- Automatic expansion if no results
- Efficient pagination handling
"""
client = get_client()
try:
if name == "get_party":
result = client.get_party(arguments["party_id"])
elif name == "get_document":
result = client.get_document(arguments["document_id"])
elif name == "get_project":
result = client.get_project(arguments["project_id"])
elif name == "get_contact":
result = client.get_contact(arguments["contact_id"])
elif name == "get_party_contacts":
result = client.get_party_contacts(arguments["party_id"])
elif name == "get_party_compliance_profile":
result = client.get_party_compliance_profile(arguments["party_id"])
elif name == "get_party_compliance_certificate":
result = client.get_party_compliance_certificate(arguments["party_id"])
elif name == "get_party_document_request":
result = client.get_party_document_request(arguments["party_id"])
elif name == "get_compliance_profile":
result = client.get_compliance_profile(arguments["profile_id"])
elif name == "get_report":
result = client.get_report(arguments["report_id"])
elif name == "list_parties":
result = _smart_list_search(client, client.get_parties, arguments)
elif name == "list_documents":
result = _smart_list_search(client, client.get_documents, arguments)
elif name == "list_projects":
result = _smart_list_search(client, client.get_projects, arguments)
elif name == "list_tags":
result = client.get_tags()
elif name == "list_compliance_profiles":
params = _build_query_params(arguments)
result = client.get_compliance_profiles(params)
elif name == "list_party_types":
result = client.get_party_types()
elif name == "list_custom_fields":
result = client.get_custom_fields()
elif name == "list_reports":
params = _build_query_params(arguments)
result = client.get_reports(params)
elif name == "list_webhooks":
params = _build_query_params(arguments)
result = client.get_webhooks(params)
elif name == "get_branding":
result = client.get_branding_item(arguments["branding_id"])
elif name == "list_branding":
params = _build_query_params(arguments)
result = client.get_branding(params)
else:
raise ValueError(f"Unknown tool: {name}")
return [TextContent(type="text", text=json.dumps(result, indent=2))]
except Exception as e:
error_msg = f"Error calling tool {name}: {str(e)}"
return [TextContent(type="text", text=json.dumps({"error": error_msg}, indent=2))]
