"""MCP server for TrustLayer API."""
import asyncio
import logging
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool, TextContent
from src.resources import (
list_parties,
list_documents,
list_projects,
list_tags,
list_compliance_profiles,
list_party_types,
list_custom_fields,
list_reports,
list_webhooks,
list_branding,
read_resource,
)
from src.tools import get_tools, call_tool
# Configure logging to stderr (MCP uses stdin/stdout for protocol)
logging.basicConfig(
level=logging.INFO, # Show info level for debugging
format='%(name)s - %(levelname)s - %(message)s',
stream=__import__('sys').stderr
)
logger = logging.getLogger(__name__)
# Create MCP server
server = Server("trustlayer-mcp")
@server.list_resources()
async def handle_list_resources() -> list[Resource]:
"""List all available resources."""
all_resources = []
# List each resource type independently, so errors in one don't block others
resource_functions = [
("parties", list_parties),
("documents", list_documents),
("projects", list_projects),
("tags", list_tags),
("compliance_profiles", list_compliance_profiles),
("party_types", list_party_types),
("custom_fields", list_custom_fields),
("reports", list_reports),
("webhooks", list_webhooks),
("branding", list_branding),
]
for name, func in resource_functions:
try:
resources = await func()
all_resources.extend(resources)
logger.info(f"Loaded {len(resources)} {name} resources")
except Exception as e:
logger.warning(f"Error loading {name} resources: {e}")
# Continue with other resources even if one fails
logger.info(f"Total resources loaded: {len(all_resources)}")
return all_resources
@server.read_resource()
async def handle_read_resource(uri: str) -> str:
"""Read a resource by URI."""
try:
return await read_resource(uri)
except Exception as e:
logger.error(f"Error reading resource {uri}: {e}")
raise
@server.list_tools()
async def handle_list_tools() -> list[Tool]:
"""List all available tools."""
try:
tools = get_tools()
logger.info(f"Returning {len(tools)} tools to client")
return tools
except Exception as e:
logger.error(f"Error listing tools: {e}", exc_info=True)
return []
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Call a tool by name with arguments."""
try:
return await call_tool(name, arguments)
except Exception as e:
logger.error(f"Error calling tool {name}: {e}")
raise
async def main():
"""Main entry point."""
try:
# Pre-load tools to verify they're available
tools = get_tools()
logger.warning(f"MCP server starting with {len(tools)} tools available")
async with stdio_server() as (read_stream, write_stream):
logger.warning("MCP server connected via stdio")
init_options = server.create_initialization_options()
logger.warning(f"Initialization options: {init_options}")
await server.run(
read_stream,
write_stream,
init_options,
)
except Exception as e:
logger.error(f"Fatal error in MCP server: {e}", exc_info=True)
raise
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Failed to start server: {e}", exc_info=True)
raise
