tools.py•4.85 kB
"""Tool handler functions for NIX MCP Server - Simplified version"""
import json
import logging
from typing import Any, Dict, List, Optional
from mcp.types import TextContent
from .abi_fetcher import ABIFetcher
from .simple_client import SimpleNixClient
logger = logging.getLogger(__name__)
async def handle_list_queries(
client: SimpleNixClient,
contract: str = "nix.q",
filter_pattern: Optional[str] = None,
environment: str = "dev"
) -> List[TextContent]:
"""
List all available NIX query actions
Args:
client: SimpleNixClient instance
contract: Contract to list queries from (default: nix.q)
filter_pattern: Optional filter pattern for query names
environment: Environment name
Returns:
List containing TextContent with available queries
"""
try:
fetcher = ABIFetcher(nodeos_api=client.nodeos_api, environment=environment)
actions = fetcher.get_actions(contract)
# Apply filter if provided
if filter_pattern:
filter_lower = filter_pattern.lower()
actions = [a for a in actions if filter_lower in a.lower()]
result = {
"contract": contract,
"environment": environment,
"total_queries": len(actions),
"queries": actions
}
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except Exception as e:
logger.error(f"Error listing queries: {e}")
return [TextContent(
type="text",
text=f"Error: {str(e)}"
)]
async def handle_get_query_abi(
client: SimpleNixClient,
query_name: str,
contract: str = "nix.q",
include_example: bool = True,
environment: str = "dev"
) -> List[TextContent]:
"""
Get ABI structure and JSON template for a specific query
Args:
client: SimpleNixClient instance
query_name: Name of the query action
contract: Contract name (default: nix.q)
include_example: Include JSON example based on common patterns
environment: Environment name
Returns:
List containing TextContent with query ABI and example
"""
try:
fetcher = ABIFetcher(nodeos_api=client.nodeos_api, environment=environment)
# Get complete resolved schema using the new resolver
action_info = fetcher.get_action_schema(contract, query_name)
if not action_info:
raise ValueError(f"Query '{query_name}' not found in contract '{contract}'")
result = {
"query": query_name,
"contract": contract,
"environment": environment,
"action_type": action_info.get("type", "unknown"),
"template": action_info.get("template", {}),
"schema": action_info.get("schema", {}),
"description": action_info.get("description", "")
}
# Add example if requested
if include_example:
result["example"] = action_info.get("example", {})
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
except Exception as e:
logger.error(f"Error getting query ABI: {e}")
return [TextContent(
type="text",
text=f"Error: {str(e)}"
)]
async def handle_query(
client: SimpleNixClient,
action: str,
params: Optional[Dict[str, Any]] = None,
contract: str = "nix.q",
environment: str = "dev"
) -> List[TextContent]:
"""
Execute a NIX query with JSON parameters
Args:
client: SimpleNixClient instance
action: Query action name
params: JSON parameters for the query
contract: Contract name (default: nix.q)
environment: Environment name
Returns:
List containing TextContent with query result
"""
try:
# Execute the query
result = await client.query(
contract=contract,
action=action,
params=params or {}
)
# Handle both JSON and raw text responses
if isinstance(result, dict):
# Add environment info to result if it's a dict
result["environment"] = environment
return [TextContent(
type="text",
text=json.dumps(result, indent=2)
)]
else:
# Return raw text response
return [TextContent(
type="text",
text=str(result)
)]
except Exception as e:
logger.error(f"Error executing query: {e}")
return [TextContent(
type="text",
text=f"Error: {str(e)}"
)]