server_fastmcp.py•4.04 kB
"""NIX MCP Server - Simplified version using JSON and ABI"""
import logging
from pathlib import Path
from typing import Any, Dict, Optional
from mcp.server import FastMCP
from pydantic import Field
from dotenv import load_dotenv
from .simple_client import SimpleNixClient
from .abi_fetcher import ABIFetcher
from .tools import (
handle_list_queries,
handle_get_query_abi,
handle_query
)
# Load environment variables
load_dotenv()
# Get logger - inherits configuration from main.py
logger = logging.getLogger(__name__)
# Initialize FastMCP server
mcp = FastMCP("nix-mcp")
async def initialize_abi_cache():
"""Initialize ABI cache on server startup"""
try:
cache_dir = Path(".abi_cache")
cache_dir.mkdir(exist_ok=True)
# Use cdev environment for initial cache as specified in expected flow
logger.info("Initializing ABI cache with cdev environment...")
try:
fetcher = ABIFetcher(environment="cdev")
# Cache the main query contract ABI
contracts = ["nix.q", "nix"]
for contract in contracts:
try:
fetcher.cache_abi(contract)
logger.info(f"Cached ABI for {contract}")
except Exception as e:
logger.warning(f"Could not cache ABI for {contract}: {e}")
logger.info("ABI cache initialization completed")
except Exception as e:
# Non-fatal - server can still run without cached ABIs
logger.warning(f"Could not initialize ABI cache (non-fatal): {e}")
except Exception as e:
# Non-fatal - server can still run without cache directory
logger.warning(f"Failed to create cache directory (non-fatal): {e}")
# Note: FastMCP doesn't have startup hooks yet
# Cache initialization will happen on first query if needed
@mcp.tool()
async def list_queries(
contract: str = Field(default="nix.q", description="Contract to list queries from"),
filter_pattern: Optional[str] = Field(default=None, description="Optional filter pattern for query names"),
environment: str = Field(default="dev", description="Environment (dev, uat, cdev, perf, simnext, prod, local)")
) -> str:
"""List all available NIX query actions from the contract ABI"""
# Create client with specified environment
client = SimpleNixClient(environment=environment)
result = await handle_list_queries(client, contract, filter_pattern, environment)
return result[0].text
@mcp.tool()
async def get_query_abi(
query_name: str = Field(..., description="Name of the query action"),
contract: str = Field(default="nix.q", description="Contract name"),
include_example: bool = Field(default=True, description="Include JSON example"),
environment: str = Field(default="dev", description="Environment (dev, uat, cdev, perf, simnext, prod, local)")
) -> str:
"""Get ABI structure and JSON template for a specific query"""
# Create client with specified environment
client = SimpleNixClient(environment=environment)
result = await handle_get_query_abi(client, query_name, contract, include_example, environment)
return result[0].text
@mcp.tool()
async def query(
action: str = Field(..., description="Query action name"),
params: Optional[Dict[str, Any]] = Field(default=None, description="JSON parameters for the query"),
contract: str = Field(default="nix.q", description="Contract name"),
environment: str = Field(default="dev", description="Environment (dev, uat, cdev, perf, simnext, prod, local)")
) -> str:
"""Execute a NIX query with JSON parameters"""
# Create client with specified environment
client = SimpleNixClient(environment=environment)
result = await handle_query(client, action, params, contract, environment)
return result[0].text
def create_server():
"""Create and return the MCP server instance"""
return mcp
if __name__ == "__main__":
mcp.run("stdio")