mcp_server_service.py

Overview

mcp_server_service.py defines the MCPServerService class, which is responsible for managing database operations related to MCP (Multi-Cloud Platform) servers within the InfiniFlow system. This service class extends a common base service (CommonService) to provide specialized methods for retrieving MCP server data based on tenant context, filtering criteria, and pagination requirements.

The primary functionality includes:

Fetching a list of MCP servers associated with a specific tenant, with support for filtering, sorting, and pagination.
Retrieving a specific MCP server by its name and tenant ID, useful for validation or lookup purposes.

This service acts as the data access layer for MCP server entities, abstracting the database queries and providing a clear API for other parts of the system to interact with MCP server records.

Class: MCPServerService

class MCPServerService(CommonService):

Description

MCPServerService is a subclass of CommonService tailored to handle operations on the MCPServer database model. It centralizes MCP server-related queries and encapsulates the logic required to manipulate MCP server data.

Attributes

model (MCPServer): The Peewee ORM model representing the MCP server table in the database. This attribute defines the entity on which the service operates.

Methods

`get_servers`

@classmethod
@DB.connection_context()
def get_servers(cls, tenant_id: str, id_list: list[str] | None, page_number, items_per_page, orderby, desc, keywords)

Description

Fetches MCP server records for a given tenant, supporting optional filtering by server IDs and search keywords. Supports sorting by any specified field and pagination of results.

Parameters

tenant_id (str): The identifier of the tenant whose MCP servers are to be retrieved.
id_list (list[str] | None): Optional list of MCP server IDs to filter the results. If None, this filter is ignored.
page_number (int or None): The page number for paginated results. If not provided, pagination is not applied.
items_per_page (int or None): Number of items per page for pagination.
orderby (str): The field name by which to order the results.
desc (bool): If True, results are ordered in descending order; ascending if False.
keywords (str or None): Search keyword to filter MCP servers by name (case-insensitive substring match). If None or empty, this filter is ignored.

Returns

list[dict] | None: A list of dictionaries, each representing an MCP server with selected fields suitable for listing. Returns None if no MCP servers match the criteria.

Usage Example

servers = MCPServerService.get_servers(
    tenant_id="tenant_123",
    id_list=None,
    page_number=1,
    items_per_page=10,
    orderby="create_date",
    desc=True,
    keywords="prod"
)
if servers:
    for server in servers:
        print(server["name"], server["url"])
else:
    print("No MCP servers found.")

Implementation Details

Uses Peewee ORM to construct a query selecting specific fields relevant for list views.
Applies filters conditionally:
- Filters by tenant ID always.
- Filters by a list of IDs if provided.
- Filters by case-insensitive keyword matching on the server name.
Orders the results by a specified field in ascending or descending order.
Applies pagination with Peewee's paginate() if page and item counts are provided.
Returns None when no matching records are found.

`get_by_name_and_tenant`

@classmethod
@DB.connection_context()
def get_by_name_and_tenant(cls, name: str, tenant_id: str)

Description

Retrieves a single MCP server record by its name within a specific tenant context. Useful for checking existence or fetching the server details by name.

Parameters

name (str): The MCP server's name to search for.
tenant_id (str): The tenant ID to scope the search.

Returns

tuple(bool, MCPServer | None):
- The first element is True if a matching MCP server exists; otherwise False.
- The second element is the MCPServer instance if found, else None.

Usage Example

exists, server = MCPServerService.get_by_name_and_tenant("server_01", "tenant_123")
if exists:
    print(f"Server found: {server.url}")
else:
    print("Server not found.")

Implementation Details

Uses the model's query method (assumed custom or inherited ORM method) to filter by name and tenant ID.
Catches any exceptions during the query and returns (False, None) on failure.

Important Implementation Details

The class uses the peewee ORM for database interactions.
Database connections are managed using the @DB.connection_context() decorator, ensuring proper connection handling.
The database model MCPServer is imported from api.db.db_models.
In get_servers, dynamic ordering is achieved by calling getter_by(orderby) on the model, which is assumed to return the corresponding field attribute.
Filtering by keywords is case-insensitive via fn.LOWER and .contains().
Pagination uses Peewee's built-in paginate() method.
The service class relies on CommonService as its base, which likely provides shared database operation utilities (not shown here).

Interaction with Other Parts of the System

Database Models: Uses MCPServer model from api.db.db_models, which maps to the MCP server database table.
Common Service Layer: Inherits common database service functionality from CommonService in api.db.services.common_service.
Database Connection: Manages database connection lifecycle via DB.connection_context().
API Layer: Likely called by API controllers or REST endpoints to fetch MCP server data for UI display or validation.
Tenant Management: The tenant ID parameter enforces multi-tenant data isolation.
Search & Pagination: Provides filtered and paginated data for potentially large MCP server lists.

Class Diagram

classDiagram
    class CommonService {
        <<abstract>>
        +query(...)
        +create(...)
        +update(...)
        +delete(...)
    }

    class MCPServerService {
        +model = MCPServer
        +get_servers(tenant_id: str, id_list: list[str] | None, page_number, items_per_page, orderby, desc, keywords)
        +get_by_name_and_tenant(name: str, tenant_id: str)
    }

    MCPServerService --|> CommonService
    MCPServerService ..> MCPServer : uses

Summary

mcp_server_service.py provides a focused service layer for MCP server entities in a multi-tenant environment. It abstracts direct database queries, supporting filtered retrieval, sorting, and pagination of MCP server records, along with existence checks by name and tenant. Its integration with the Peewee ORM and the common service base class enables consistent and efficient database operations for the MCP server domain within the InfiniFlow application.