"""MCP resources for TrustLayer entities."""
from mcp.types import Resource
from typing import List, Dict, Any
import json
import logging
logger = logging.getLogger(__name__)
from src.trustlayer_client import TrustLayerClient
from src.config import settings, api_token_context
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,
)
async def list_parties() -> List[Resource]:
"""List all parties (automatically fetches all pages)."""
client = get_client()
# Automatically fetch all pages for resources
response = client.get_parties(fetch_all=True)
parties = response.get("data", [])
return [
Resource(
uri=f"trustlayer://parties/{party['id']}",
name=party.get("name", party["id"]),
description=f"Party: {party.get('name', 'N/A')}",
mimeType="application/json",
)
for party in parties
]
async def list_documents() -> List[Resource]:
"""List all documents (automatically fetches all pages)."""
client = get_client()
# Automatically fetch all pages for resources
response = client.get_documents(fetch_all=True)
documents = response.get("data", [])
return [
Resource(
uri=f"trustlayer://documents/{doc['id']}",
name=doc.get("name", doc["id"]),
description=f"Document: {doc.get('name', 'N/A')}",
mimeType="application/json",
)
for doc in documents
]
async def list_projects() -> List[Resource]:
"""List all projects (automatically fetches all pages)."""
client = get_client()
# Automatically fetch all pages for resources
response = client.get_projects(fetch_all=True)
projects = response.get("data", [])
return [
Resource(
uri=f"trustlayer://projects/{project['id']}",
name=project.get("name", project["id"]),
description=f"Project: {project.get('name', 'N/A')}",
mimeType="application/json",
)
for project in projects
]
async def list_contacts() -> List[Resource]:
"""List all contacts."""
client = get_client()
# Contacts are typically accessed through parties
# For now, we'll return an empty list or fetch from all parties
return []
async def list_tags() -> List[Resource]:
"""List all tags."""
client = get_client()
response = client.get_tags()
tags = response.get("data", [])
return [
Resource(
uri=f"trustlayer://tags/{tag['id']}",
name=tag.get("name", tag["id"]),
description=f"Tag: {tag.get('name', 'N/A')}",
mimeType="application/json",
)
for tag in tags
]
async def list_compliance_profiles() -> List[Resource]:
"""List all compliance profiles."""
client = get_client()
response = client.get_compliance_profiles()
profiles = response.get("data", [])
return [
Resource(
uri=f"trustlayer://compliance-profiles/{profile['id']}",
name=profile.get("name", profile["id"]),
description=f"Compliance Profile: {profile.get('name', 'N/A')}",
mimeType="application/json",
)
for profile in profiles
]
async def list_party_types() -> List[Resource]:
"""List all party types."""
client = get_client()
response = client.get_party_types()
party_types = response.get("data", [])
return [
Resource(
uri=f"trustlayer://party-types/{pt['id']}",
name=pt.get("name", pt["id"]),
description=f"Party Type: {pt.get('name', 'N/A')}",
mimeType="application/json",
)
for pt in party_types
]
async def list_custom_fields() -> List[Resource]:
"""List all custom fields."""
client = get_client()
response = client.get_custom_fields()
fields = response.get("data", [])
return [
Resource(
uri=f"trustlayer://custom-fields/{field['id']}",
name=field.get("name", field["id"]),
description=f"Custom Field: {field.get('name', 'N/A')}",
mimeType="application/json",
)
for field in fields
]
async def list_reports() -> List[Resource]:
"""List all reports."""
client = get_client()
response = client.get_reports()
reports = response.get("data", [])
return [
Resource(
uri=f"trustlayer://reports/{report['id']}",
name=report.get("name", report["id"]),
description=f"Report: {report.get('name', 'N/A')}",
mimeType="application/json",
)
for report in reports
]
async def list_webhooks() -> List[Resource]:
"""List all webhooks."""
client = get_client()
response = client.get_webhooks()
webhooks = response.get("data", [])
return [
Resource(
uri=f"trustlayer://webhooks/{webhook['id']}",
name=webhook.get("url", webhook["id"]),
description=f"Webhook: {webhook.get('url', 'N/A')}",
mimeType="application/json",
)
for webhook in webhooks
]
async def list_branding() -> List[Resource]:
"""List all branding."""
try:
client = get_client()
response = client.get_branding()
branding_items = response.get("data", [])
return [
Resource(
uri=f"trustlayer://branding/{item['id']}",
name=item.get("name", item["id"]),
description=f"Branding: {item.get('name', 'N/A')}",
mimeType="application/json",
)
for item in branding_items
]
except Exception as e:
# Branding endpoint may return 403 for read-only tokens
logger.warning(f"Could not list branding (may require higher permissions): {e}")
return []
async def read_resource(uri: str) -> str:
"""Read resource content by URI."""
client = get_client()
if uri.startswith("trustlayer://parties/"):
party_id = uri.replace("trustlayer://parties/", "")
data = client.get_party(party_id)
return json.dumps(data, indent=2)
elif uri.startswith("trustlayer://documents/"):
doc_id = uri.replace("trustlayer://documents/", "")
data = client.get_document(doc_id)
return json.dumps(data, indent=2)
elif uri.startswith("trustlayer://projects/"):
project_id = uri.replace("trustlayer://projects/", "")
data = client.get_project(project_id)
return json.dumps(data, indent=2)
elif uri.startswith("trustlayer://tags/"):
tag_id = uri.replace("trustlayer://tags/", "")
# Tags don't have individual GET endpoint, return from list
response = client.get_tags()
tags = response.get("data", [])
tag = next((t for t in tags if t["id"] == tag_id), None)
if tag:
return json.dumps(tag, indent=2)
raise ValueError(f"Tag {tag_id} not found")
elif uri.startswith("trustlayer://compliance-profiles/"):
profile_id = uri.replace("trustlayer://compliance-profiles/", "")
data = client.get_compliance_profile(profile_id)
return json.dumps(data, indent=2)
elif uri.startswith("trustlayer://party-types/"):
# Party types don't have individual GET endpoint
response = client.get_party_types()
party_types = response.get("data", [])
pt_id = uri.replace("trustlayer://party-types/", "")
pt = next((p for p in party_types if p["id"] == pt_id), None)
if pt:
return json.dumps(pt, indent=2)
raise ValueError(f"Party type {pt_id} not found")
elif uri.startswith("trustlayer://custom-fields/"):
# Custom fields don't have individual GET endpoint
response = client.get_custom_fields()
fields = response.get("data", [])
field_id = uri.replace("trustlayer://custom-fields/", "")
field = next((f for f in fields if f["id"] == field_id), None)
if field:
return json.dumps(field, indent=2)
raise ValueError(f"Custom field {field_id} not found")
elif uri.startswith("trustlayer://reports/"):
report_id = uri.replace("trustlayer://reports/", "")
data = client.get_report(report_id)
return json.dumps(data, indent=2)
elif uri.startswith("trustlayer://webhooks/"):
# Webhooks don't have individual GET endpoint
response = client.get_webhooks()
webhooks = response.get("data", [])
webhook_id = uri.replace("trustlayer://webhooks/", "")
webhook = next((w for w in webhooks if w["id"] == webhook_id), None)
if webhook:
return json.dumps(webhook, indent=2)
raise ValueError(f"Webhook {webhook_id} not found")
elif uri.startswith("trustlayer://branding/"):
branding_id = uri.replace("trustlayer://branding/", "")
data = client.get_branding_item(branding_id)
return json.dumps(data, indent=2)
else:
raise ValueError(f"Unknown resource URI: {uri}")
