AACT Clinical Trials MCP Server
by navisbio
- src
- mcp_server_aact
import logging
from typing import Any, Optional
import mcp.types as types
from mcp_server_aact.database import AACTDatabase
from mcp_server_aact.memo_manager import MemoManager
from mcp_server_aact.errors import ToolError, handle_errors
logger = logging.getLogger('mcp_aact_server.tools')
class ToolManager:
def __init__(self, db: AACTDatabase, memo_manager: MemoManager):
self.db = db
self.memo_manager = memo_manager
logger.info("ToolManager initialized")
def get_available_tools(self) -> list[types.Tool]:
"""Return list of available tools"""
logger.debug("Retrieving available tools")
tools = [
types.Tool(
name="read-query",
description="Execute a SELECT query on the AACT clinical trials database. Use this tool to extract and analyze specific data from any table.",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "SELECT SQL query to execute"},
},
"required": ["query"],
},
),
types.Tool(
name="list-tables",
description="Get an overview of all available tables in the AACT database. This tool helps you understand the database structure before starting your analysis to identify relevant data sources.",
inputSchema={
"type": "object",
"properties": {},
},
),
types.Tool(
name="describe-table",
description="Examine the detailed structure of a specific AACT table, including column names and data types. Use this before querying to ensure you target the right columns and understand the data format.",
inputSchema={
"type": "object",
"properties": {
"table_name": {"type": "string", "description": "Name of the table to describe"},
},
"required": ["table_name"],
},
),
types.Tool(
name="append-insight",
description="Record key findings and insights discovered during your analysis. Use this tool whenever you uncover meaningful patterns, trends, or notable observations about clinical trials. This helps build a comprehensive analytical narrative and ensures important discoveries are documented.",
inputSchema={
"type": "object",
"properties": {
"finding": {"type": "string", "description": "Analysis finding about trial patterns or trends"},
},
"required": ["finding"],
},
),
]
logger.debug(f"Retrieved {len(tools)} available tools")
return tools
@handle_errors(ToolError, "Error executing tool {name}: {error}")
async def execute_tool(self, name: str, arguments: dict[str, Any] | None) -> list[types.TextContent]:
"""Execute a tool with given arguments"""
logger.info(f"Executing tool: {name} with arguments: {arguments}")
if name not in {tool.name for tool in self.get_available_tools()}:
raise ToolError(f"Unknown tool: {name}")
if not arguments and name != "list-tables":
raise ToolError("Missing required arguments")
if name == "list-tables":
logger.debug("Executing list-tables query")
results = self.db.execute_query("""
SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'ctgov'
ORDER BY table_name;
""")
logger.info(f"Retrieved {len(results)} tables")
return [types.TextContent(type="text", text=str(results))]
elif name == "describe-table":
if "table_name" not in arguments:
raise ToolError("Missing table_name argument")
logger.debug(f"Describing table: {arguments['table_name']}")
results = self.db.execute_query("""
SELECT column_name, data_type, character_maximum_length
FROM information_schema.columns
WHERE table_schema = 'ctgov'
AND table_name = %s
ORDER BY ordinal_position;
""", {"table_name": arguments["table_name"]})
logger.info(f"Retrieved {len(results)} columns for table {arguments['table_name']}")
return [types.TextContent(type="text", text=str(results))]
elif name == "read-query":
query = arguments.get("query", "").strip()
logger.debug(f"Executing query: {query}")
results = self.db.execute_query(query)
logger.info(f"Query returned {len(results)} rows")
return [types.TextContent(type="text", text=str(results))]
elif name == "append-insight":
if "finding" not in arguments:
raise ToolError("Missing finding argument")
logger.debug(f"Adding insight: {arguments['finding'][:50]}...")
self.memo_manager.add_insights(arguments["finding"])
logger.info("Landscape finding added successfully")
return [types.TextContent(type="text", text="Insight added")]
raise ToolError(f"Unhandled tool: {name}")