MCP Server

grizabella.mcp.server

Grizabella MCP Server.

This module provides an MCP (Model Context Protocol) server for Grizabella, exposing its core functionalities as tools that can be called remotely. It uses FastMCP to define and serve these tools.

Server Description: This MCP server exposes the core functionalities of the Grizabella knowledge management system, allowing for the creation, retrieval, and querying of structured data objects and their relationships.

get_grizabella_client()

Returns the shared Grizabella client instance.

Source code in grizabella/mcp/server.py

def get_grizabella_client() -> Grizabella:
    """Returns the shared Grizabella client instance."""
    if grizabella_client_instance is None:
        raise GrizabellaException("Grizabella client is not initialized.")
    return grizabella_client_instance

get_grizabella_db_path(db_path_arg=None)

Determines the database path from arg, env var, or default.

Source code in grizabella/mcp/server.py

def get_grizabella_db_path(db_path_arg: Optional[str] = None) -> Union[str, Path]:
    """Determines the database path from arg, env var, or default."""
    if db_path_arg:
        return db_path_arg
    return os.getenv(GRIZABELLA_DB_PATH_ENV_VAR, DEFAULT_GRIZABELLA_DB_PATH)

main()

Initializes client and runs the FastMCP application.

Source code in grizabella/mcp/server.py

def main():
    """Initializes client and runs the FastMCP application."""
    parser = argparse.ArgumentParser(description="Grizabella MCP Server")
    parser.add_argument("--db-path", help="Path to the Grizabella database.")
    args = parser.parse_args()

    global grizabella_client_instance
    db_path = get_grizabella_db_path(args.db_path)

    with Grizabella(db_name_or_path=db_path, create_if_not_exists=True) as gb:
        grizabella_client_instance = gb
        app.run()

mcp_find_objects(type_name, filter_criteria=None, limit=None) async

Finds and retrieves a list of objects of a given type, with optional filtering criteria.

Source code in grizabella/mcp/server.py

@app.tool(
    name="find_objects",
    description=(
        "Finds and retrieves a list of objects of a given type, with optional filtering criteria.\n\n"
        "Example:\n"
        "To find all 'Person' objects where the age is greater than 30:\n"
        '{\n'
        '  "args": {\n'
        '    "type_name": "Person",\n'
        '    "filter_criteria": {\n'
        '      "age": {">": 30}\n'
        '    },\n'
        '    "limit": 10\n'
        '  }\n'
        '}'
    ),
)
async def mcp_find_objects(
    type_name: str,
    filter_criteria: Optional[dict[str, Any]] = None,
    limit: Optional[int] = None,
) -> list[ObjectInstance]:
    """
    Finds and retrieves a list of objects of a given type, with optional filtering criteria.
    """
    # ctx: ToolContext,
    try:
        gb = get_grizabella_client()
        return gb.find_objects(
            type_name=type_name,
            filter_criteria=filter_criteria,
            limit=limit,
        )
    except GrizabellaException as e:
        msg = f"MCP: Error finding objects of type '{type_name}': {e}"
        raise GrizabellaException(msg) from e
    except Exception as e:  # pylint: disable=broad-except
        msg = f"MCP: Unexpected error finding objects of type '{type_name}': {e}"
        raise Exception(msg) from e

mcp_get_embedding_vector_for_text(args) async

Generates an embedding vector for a given text using a specified embedding definition.

Source code in grizabella/mcp/server.py

@app.tool(
    name="get_embedding_vector_for_text",
    description="Generates an embedding vector for a given text using a specified embedding definition.",
)
async def mcp_get_embedding_vector_for_text(args: GetEmbeddingVectorForTextArgs) -> EmbeddingVector:
    """Generates an embedding vector for a given text using a specified embedding definition."""
    gb = get_grizabella_client()
    temp_obj_id = uuid.uuid4()
    embedding_def = gb.get_embedding_definition(args.embedding_definition_name)
    try:
        # 1. Get the embedding definition
        if not embedding_def:
            raise GrizabellaException(f"Embedding definition '{args.embedding_definition_name}' not found.")

        # 2. Get the corresponding object type definition
        obj_type_def = gb.get_object_type_definition(embedding_def.object_type_name)
        if not obj_type_def:
            raise GrizabellaException(f"Object type '{embedding_def.object_type_name}' not found for embedding definition.")

        # 3. Create a temporary object with dummy data for required fields
        temp_properties = {}
        for prop_def in obj_type_def.properties:
            if prop_def.name == embedding_def.source_property_name:
                temp_properties[prop_def.name] = args.text_to_embed
            elif not prop_def.is_nullable:
                # Provide dummy data for non-nullable fields
                if prop_def.data_type == PropertyDataType.TEXT:
                    temp_properties[prop_def.name] = f"dummy_{prop_def.name}"
                elif prop_def.data_type == PropertyDataType.INTEGER:
                    temp_properties[prop_def.name] = 0
                elif prop_def.data_type == PropertyDataType.FLOAT:
                    temp_properties[prop_def.name] = 0.0
                elif prop_def.data_type == PropertyDataType.BOOLEAN:
                    temp_properties[prop_def.name] = False
                elif prop_def.data_type == PropertyDataType.DATETIME:
                    temp_properties[prop_def.name] = datetime.now(timezone.utc).isoformat()
                elif prop_def.data_type == PropertyDataType.UUID:
                    temp_properties[prop_def.name] = str(uuid.uuid4())
                else:
                    # For BLOB, JSON, etc., we might need more robust dummy data generation
                    temp_properties[prop_def.name] = None # This might fail if not nullable, but it's a start

        temp_obj_instance = ObjectInstance(
            id=temp_obj_id,
            object_type_name=embedding_def.object_type_name,
            properties=temp_properties
        )
        gb.upsert_object(temp_obj_instance)

        # 4. Retrieve the embedding vector
        # This relies on internal access, similar to the original test.
        embedding_instances = gb._db_manager.lancedb_adapter.get_embedding_instances_for_object(
           object_instance_id=temp_obj_id,
           embedding_definition_name=args.embedding_definition_name
        )

        if not embedding_instances:
            raise GrizabellaException("Failed to generate or retrieve embedding for temporary object.")

        vector = embedding_instances[0].vector
        if not vector:
            raise GrizabellaException("Retrieved embedding instance has no vector.")

        return EmbeddingVector(vector=vector)

    finally:
        # 5. Clean up the temporary object
        if embedding_def:
            try:
                gb.delete_object(object_id=str(temp_obj_id), type_name=embedding_def.object_type_name)
            except Exception as e:
                # Log cleanup error but don't let it hide the main result/error
                print(f"MCP: Warning: Failed to clean up temporary object {temp_obj_id}: {e}", file=sys.stderr)

mcp_get_incoming_relations(object_id, type_name, relation_type_name=None) async

Retrieves all incoming relations to a specific object.

Source code in grizabella/mcp/server.py

@app.tool(
    name="get_incoming_relations",
    description=(
        "Retrieves all incoming relations to a specific object.\n\n"
        "Example:\n"
        "To get all incoming relations to Jane Doe's 'Person' object:\n"
        '{\n'
        '  "args": {\n'
        '    "object_id": "jane_doe_456",\n'
        '    "type_name": "Person"\n'
        '  }\n'
        '}'
    ),
)
async def mcp_get_incoming_relations(
    object_id: str, type_name: str, relation_type_name: Optional[str] = None
) -> list[RelationInstance]:
    """
    Retrieves all incoming relations to a specific object.
    """
    # ctx: ToolContext,
    try:
        gb = get_grizabella_client()
        return gb.get_incoming_relations(
            object_id=object_id,
            type_name=type_name,
            relation_type_name=relation_type_name,
        )
    except GrizabellaException as e:
        msg = f"MCP: Error getting incoming relations for object '{object_id}': {e}"
        raise GrizabellaException(msg) from e
    except Exception as e:  # pylint: disable=broad-except
        msg = f"MCP: Unexpected error getting incoming relations for object '{object_id}': {e}"
        raise Exception(msg) from e

mcp_get_outgoing_relations(object_id, type_name, relation_type_name=None) async

Retrieves all outgoing relations from a specific object.

Source code in grizabella/mcp/server.py

@app.tool(
    name="get_outgoing_relations",
    description=(
        "Retrieves all outgoing relations from a specific object.\n\n"
        "Example:\n"
        "To get all outgoing relations from John Doe's 'Person' object:\n"
        '{\n'
        '  "args": {\n'
        '    "object_id": "john_doe_123",\n'
        '    "type_name": "Person"\n'
        '  }\n'
        '}'
    ),
)
async def mcp_get_outgoing_relations(
    object_id: str, type_name: str, relation_type_name: Optional[str] = None
) -> list[RelationInstance]:
    """
    Retrieves all outgoing relations from a specific object.
    """
    # ctx: ToolContext,
    try:
        gb = get_grizabella_client()
        return gb.get_outgoing_relations(
            object_id=object_id,
            type_name=type_name,
            relation_type_name=relation_type_name,
        )
    except GrizabellaException as e:
        msg = f"MCP: Error getting outgoing relations for object '{object_id}': {e}"
        raise GrizabellaException(msg) from e
    except Exception as e:  # pylint: disable=broad-except
        msg = f"MCP: Unexpected error getting outgoing relations for object '{object_id}': {e}"
        raise Exception(msg) from e

mcp_search_similar_objects(object_id, type_name, n_results=5, search_properties=None) async

Searches for objects that are semantically similar to a given object, based on embeddings of their properties.

Source code in grizabella/mcp/server.py

@app.tool(
    name="search_similar_objects",
    description=(
        "Searches for objects that are semantically similar to a given object, based on embeddings "
        "of their properties. Note: This feature is not yet fully implemented.\n\n"
        "Example:\n"
        "To find 5 objects similar to John Doe's 'Person' object:\n"
        '{\n'
        '  "args": {\n'
        '    "object_id": "john_doe_123",\n'
        '    "type_name": "Person",\n'
        '    "n_results": 5\n'
        '  }\n'
        '}'
    ),
)
async def mcp_search_similar_objects(
    object_id: str,
    type_name: str,
    n_results: int = 5,
    search_properties: Optional[list[str]] = None,
) -> list[tuple[ObjectInstance, float]]:
    """
    Searches for objects that are semantically similar to a given object, based on embeddings of their properties.
    """
    # ctx: ToolContext,
    try:
        gb = get_grizabella_client()
        # The Grizabella client's search_similar_objects currently raises NotImplementedError.
        # We must call it to respect the interface, but handle the expected error.
        # If it were implemented, results would be List[Tuple[ObjectInstance, float]].
        # For now, to satisfy Pylint and type checkers if the method were to return,
        # we can assign and then immediately handle the expected NotImplementedError.
        # However, a cleaner approach is to directly call and handle.

        # Attempt the call and handle NotImplementedError specifically.
        # Other GrizabellaExceptions or general Exceptions will be caught below.
        try:
            # This line will raise NotImplementedError based on current client.py
            results: list[
                tuple[ObjectInstance, float]
            ] = gb.search_similar_objects(
                object_id=object_id,
                type_name=type_name,
                n_results=n_results,
                search_properties=search_properties,
            )
            return results  # This line will not be reached if NotImplementedError is raised
        except NotImplementedError as nie:
            # Specific handling for the known unimplemented feature.
            # Raising a general Exception here for the MCP layer is acceptable to signal this state.
            msg = f"MCP: search_similar_objects feature is not yet implemented in the Grizabella client: {nie}"
            raise Exception(msg) from nie

    except GrizabellaException as e:
        # Handle other Grizabella-specific errors, re-raise as GrizabellaException
        msg = f"MCP: Error searching similar objects for '{object_id}': {e}"
        raise GrizabellaException(msg) from e
    except Exception as e:  # pylint: disable=broad-except
        # Handle any other unexpected errors, re-raise as general Exception
        msg = f"MCP: Unexpected error searching similar objects for '{object_id}': {e}"
        raise Exception(msg) from e