from utils.serverlogging import log_info, log_debug
from mcp.server.fastmcp import FastMCP
from utils.schema import set_schema, get_schema
from utils.db import insert_record, update_record, get_all_records, search_records, create_database
# Create FastMCP server with path /mcp
mcp = FastMCP("tinydb-emcipi", streamable_http_path="/mcp")
# Holds the name of the database that tools should use by default if not
# explicitly provided. LLM agents can update this using switch_active_db.
_ACTIVE_DB: str | None = None
def init_mcp(run_server: bool = False):
log_info("mcp init complete.")
if run_server:
log_info(f"Starting MCP server: {mcp.name}")
log_mcp_server_details()
# Start MCP server with Uvicorn for Streamable HTTP transport
import uvicorn
app = mcp.streamable_http_app()
uvicorn.run(app, host="127.0.0.1", port=8000)
log_info("MCP server is now listening for HTTP requests on http://127.0.0.1:8000/mcp.")
# MCP protocol functions using SDK decorators
print("DEBUG: About to register list_databases_tool")
@mcp.tool(
description=(
"WHEN the user asks things like 'what data do we have', 'show databases', 'do you have any assets', or similar, "
"call this tool to list every available TinyDB database. Use this first to discover what databases exist before any specific query."
)
)
def list_databases_tool() -> dict:
from utils.serverlogging import log_info
from utils.db import list_databases
log_info("list_databases_tool called")
databases = list_databases()
return {
"status": "success",
"databases": sorted(databases),
"count": len(databases),
"tip": "Use get_database_info_tool with any database name to explore its structure."
}
print("DEBUG: list_databases_tool registered")
print("DEBUG: About to register get_database_info_tool")
@mcp.tool(
description=(
"WHEN the user says 'describe database', 'tell me about current db', 'what fields are in it', etc., call this tool. "
"Returns schema, sample records, and available fields for the ACTIVE database."
)
)
# signature changed: no db_name param
def get_database_info_tool() -> dict:
from utils.serverlogging import log_info
import os
from utils.config import get_config
from tinydb import TinyDB
if _ACTIVE_DB is None:
log_info("get_database_info_tool called but no active database is set.")
return {
"status": "error",
"message": "No active database selected. Call switch_active_db first."
}
db_name = _ACTIVE_DB
log_info(f"get_database_info_tool called for db: {db_name}")
# Get schema information
schema = get_schema(db_name)
log_info(f"get_database_info_tool queried schema {schema} for db: {db_name}")
# Get database path and check if it exists
database_dir = get_config("databasePath")
db_path = os.path.join(database_dir, f"{db_name}.json")
if not os.path.exists(db_path):
return {
"status": "success",
"database": db_name,
"schema": schema,
"sample_records": [],
"available_fields": [],
"total_records": 0,
"usage_tip": "Database file does not exist. Create records first."
}
# Open database and get just a few sample records efficiently
db = TinyDB(db_path)
total_count = len(db)
sample_records = db.all()[:3]
# Extract field names from schema and sample records
field_names_set = set()
if schema:
field_names_set.update(schema.keys())
for record in sample_records:
field_names_set.update(record.keys())
field_names = sorted(field_names_set)
# Build usage example dynamically based on actual fields
usage_example = {}
if field_names:
first_field = field_names[0]
sample_value = sample_records[0].get(first_field, "example_value") if sample_records else "example_value"
usage_example = {first_field: sample_value}
return {
"status": "success",
"database": db_name,
"schema": schema,
"sample_records": sample_records,
"available_fields": field_names,
"total_records": total_count,
"usage_tip": usage_example,
}
print("DEBUG: get_database_info_tool registered")
print("DEBUG: About to register get_all_records_tool")
@mcp.tool(
description=(
"WHEN the user requests: 'show everything', 'give me all records', 'dump the database', call this tool to return the full contents of the currently active database. "
"First set the active database with switch_active_db if none is selected."
)
)
# Changed: no longer requires db_name parameter
def get_all_records_tool() -> dict:
from utils.serverlogging import log_info
# Use the module-level _ACTIVE_DB
if _ACTIVE_DB is None:
log_info("get_all_records_tool called but no active database is set.")
return {
"status": "error",
"message": "No active database selected. Call switch_active_db first."
}
log_info(f"get_all_records_tool called for db: {_ACTIVE_DB}")
records = get_all_records(_ACTIVE_DB)
return {"status": "success", "database": _ACTIVE_DB, "records": records, "count": len(records)}
print("DEBUG: get_all_records_tool registered")
print("DEBUG: About to register search_records_tool")
@mcp.tool(
description=(
"WHEN the user says 'search', 'find', 'filter', or provides criteria like Status=Licensed, call this tool. "
"Pass a JSON object with field-value pairs, e.g. {'Status':'Licensed'}. Uses the ACTIVE database."
)
)
# signature changed: only query param
def search_records_tool(query: dict) -> dict:
from utils.serverlogging import log_info
if _ACTIVE_DB is None:
log_info("search_records_tool called but no active database is set.")
return {
"status": "error",
"message": "No active database selected. Call switch_active_db first."
}
db_name = _ACTIVE_DB
log_info(f"search_records_tool called for db: {db_name}, query: {query}")
records = search_records(db_name, query)
schema = get_schema(db_name)
return {
"status": "success",
"database": db_name,
"records": records,
"count": len(records),
"query_used": query,
"schema": schema,
"tip": "If no results found, review the schema field list above or use get_database_info_tool for samples."
}
print("DEBUG: search_records_tool registered")
print("DEBUG: About to register set_schema_tool")
@mcp.tool(
description=(
"WHEN the user wants to define or change the allowed fields/types for the ACTIVE database (e.g. 'add serial_number field'), call this tool. "
"Always ask for confirmation before running because it can overwrite validation rules."
)
)
# signature changed: only field_schema param
def set_schema_tool(field_schema: dict) -> dict:
from utils.serverlogging import log_info
if _ACTIVE_DB is None:
log_info("set_schema_tool called but no active database is set.")
return {
"status": "error",
"message": "No active database selected. Call switch_active_db first."
}
db_name = _ACTIVE_DB
log_info(f"set_schema_tool called for db: {db_name}")
set_schema(db_name, field_schema)
log_info(f"Schema for '{db_name}' updated.")
return {"status": "success", "database": db_name, "message": f"Schema for '{db_name}' updated.", "schema": field_schema}
print("DEBUG: set_schema_tool registered")
print("DEBUG: About to register add_record_tool")
@mcp.tool(
description=(
"WHEN the user requests to add / create / insert an asset or record into the ACTIVE database, call this tool immediately. "
"Provide fields matching the schema; afterwards report success so the user can adjust if needed."
)
)
# signature changed: only record param
def add_record_tool(record: dict) -> dict:
from utils.serverlogging import log_info
if _ACTIVE_DB is None:
log_info("add_record_tool called but no active database is set.")
return {
"status": "error",
"message": "No active database selected. Call switch_active_db first."
}
db_name = _ACTIVE_DB
log_info(f"add_record_tool called for db: {db_name}")
valid, error = insert_record(db_name, record)
if not valid:
log_info(f"Failed to add record to '{db_name}': {error}")
return {
"status": "error",
"message": error,
"tip": "Use get_database_info_tool to see the correct schema and field format."
}
log_info(f"Record added to '{db_name}'.")
return {"status": "success", "database": db_name, "message": f"Record added to '{db_name}'.", "record": record}
print("DEBUG: add_record_tool registered")
print("DEBUG: About to register update_record_tool")
@mcp.tool(
description=(
"WHEN the user asks to modify or correct an existing record in the ACTIVE database, call this tool with the record_id and updated fields. "
"Use get_all_records_tool first to locate the ID, then perform the update."
)
)
# signature changed: no db_name param
def update_record_tool(record_id: int, record: dict) -> dict:
from utils.serverlogging import log_info
if _ACTIVE_DB is None:
log_info("update_record_tool called but no active database is set.")
return {
"status": "error",
"message": "No active database selected. Call switch_active_db first."
}
db_name = _ACTIVE_DB
log_info(f"update_record_tool called for db: {db_name}, record_id: {record_id}")
valid, error = update_record(db_name, record_id, record)
if not valid:
log_info(f"Failed to update record {record_id} in '{db_name}': {error}")
return {
"status": "error",
"message": error,
"tip": "Use get_database_info_tool to see the correct schema and field format."
}
log_info(f"Record {record_id} updated in '{db_name}'.")
return {"status": "success", "database": db_name, "message": f"Record {record_id} updated in '{db_name}'.", "updated_record": record}
print("DEBUG: update_record_tool registered")
print("DEBUG: About to register switch_active_db_tool")
@mcp.tool(
description=(
"Call this first to select which TinyDB database subsequent tools should operate on. "
"Pass the database name (without .json). After a successful switch, you can use get_all_records_tool, search_records_tool, etc."
)
)
def switch_active_db_tool(db_name: str) -> dict:
"""Tool wrapper for switch_active_db helper."""
new_db = switch_active_db(db_name)
return {"status": "success", "active_database": new_db}
print("DEBUG: switch_active_db_tool registered")
print("DEBUG: About to register create_database_tool")
@mcp.tool(
description=(
"Proactively WHEN the user will need a brand-new TinyDB database, call this tool, or when no potentially relevant database is already existing. "
"Provide the database name (without .json). Optionally supply an initial JSON schema in the same call; "
"if provided it will be saved atomically. If not provided the next logical prompt should be defining and setting a schema."
"Afterwards the database is then ready for add_record_tool."
)
)
def create_database_tool(db_name: str, field_schema: dict | None = None) -> dict:
"""Create a database file and, if *field_schema* is given, store it immediately."""
from utils.serverlogging import log_info
create_database(db_name)
schema_saved = False
if field_schema:
set_schema(db_name, field_schema)
schema_saved = True
log_info(f"Schema saved for new DB '{db_name}'.")
# Automatically set as active DB for convenience
switch_active_db(db_name)
return {
"status": "success",
"database": db_name,
"schema_saved": schema_saved,
"tip": (
"Database created and selected. You can now add records with add_record_tool." if schema_saved
else "Database created and selected. Define a schema next with set_schema_tool or start adding records."
),
}
print("DEBUG: create_database_tool registered")
# ---------------------------------------------------------------------------
# Active database handling
# ---------------------------------------------------------------------------
def switch_active_db(db_name: str) -> str:
"""Set the global *_ACTIVE_DB* variable so subsequent tool calls can
infer which TinyDB database to operate on.
Parameters
----------
db_name : str
The logical name of the TinyDB database (without .json extension).
Returns
-------
str
The name that was set, for convenience.
"""
global _ACTIVE_DB
_ACTIVE_DB = db_name.lower().strip()
log_info(f"Active database switched to '{_ACTIVE_DB}'.")
return _ACTIVE_DB
def discover_mcp_server_details():
"""
Returns a dict of MCP server details for logging/diagnostics.
"""
# Since the list methods are async, we can't call them from a sync function
# Instead, we'll use a simpler approach to count registered items
details = {
"server_name": getattr(mcp, "name", None),
"transport": getattr(mcp.settings, "transport", None),
"host": getattr(mcp.settings, "host", None),
"port": getattr(mcp.settings, "port", None),
"debug_mode": getattr(mcp.settings, "debug", None),
"log_level": getattr(mcp.settings, "log_level", None),
"auth_enabled": hasattr(mcp, "auth"),
"tools": "async_method_available",
"resources": "async_method_available",
"prompts": "async_method_available",
}
return details
def log_mcp_server_details():
"""
Logs MCP server details at debug level if log level is debug.
"""
details = discover_mcp_server_details()
log_debug(f"MCP Server Details: {details}")
# Debug what attributes mcp actually has
log_debug(f"MCP object attributes: {[attr for attr in dir(mcp) if not attr.startswith('_')]}")
# Check for alternative tool storage
for attr in ['_tools', 'tool_registry', '_tool_registry', 'registered_tools']:
if hasattr(mcp, attr):
log_debug(f"Found {attr}: {getattr(mcp, attr)}")