Introduction
The Model Context Protocol (MCP), as defined by Anthropic, establishes a
framework for interaction between large language model (LLM) clients and
various data sources and services, known as MCP servers. In this ecosystem,
LLM clients, typically large language models or applications powered by them,
consume the capabilities offered by these servers, which can range from
providing real-time data feeds to executing complex computational tasks. As
the number and diversity of MCP servers grow, a critical challenge emerges:
how can an LLM client efficiently discover and select the most relevant MCP
servers for a given task or user query? This article addresses this challenge
by proposing and detailing the implementation of an LLM-based MCP Server
Search Engine. This engine empowers LLM clients to intelligently search for
MCP servers based on a user-defined domain, topic, or subject, by interacting
with server catalogs, directories, and libraries available on the internet. We
will explore all constituents and details of such an engine, providing a deep
dive into its architecture, implementation considerations, and a practical
running example.
I. Understanding the MCP Ecosystem
To appreciate the necessity and functionality of an LLM-based search engine,
it is crucial to first understand the fundamental components of the MCP
ecosystem.
A. MCP Clients
MCP clients are typically large language models or applications powered by
them. These clients are designed to perform a wide array of tasks, such as
answering questions, generating content, or automating workflows. To enhance
their capabilities, LLM clients often need to access external, up-to-date,
or specialized information and services that are beyond their internal
training data. This is precisely where MCP servers become invaluable. An LLM
client, for instance, might require current stock market data, weather
forecasts, or access to a proprietary internal database to fulfill a user's
request accurately and comprehensively.
B. MCP Servers
MCP servers are external services or data sources that expose their
functionality and data through a standardized protocol, making them
accessible and consumable by MCP clients. These servers act as specialized
"tools" or "knowledge bases" that an LLM can leverage. Examples of MCP
servers could include a financial data API providing real-time stock quotes,
a weather service offering local forecasts, a document management system for
retrieving specific enterprise documents, or a computational engine for
solving mathematical problems. Each MCP server typically offers a specific
set of functionalities and operates within a defined domain.
C. The Need for Discovery
Given the potential proliferation of diverse MCP servers, the sheer volume
and variety can make manual discovery and integration a significant hurdle.
An LLM client cannot inherently know about every available MCP server and
its specific capabilities. Therefore, a mechanism is required to enable
dynamic discovery, allowing the LLM client to identify and connect to the
most appropriate server based on the context of a user's request. Without an
efficient discovery mechanism, the full potential of the MCP ecosystem, where
LLMs can seamlessly integrate with external tools, remains largely untapped.
This is the core problem that an LLM-based MCP Server Search Engine aims to
solve, acting as a bridge between user intent and available MCP server
resources.
II. Architecture of an LLM-based MCP Server Search Engine
The design of an LLM-based MCP Server Search Engine involves several
interconnected components that work in concert to process user queries,
search for relevant servers, and present actionable results.
A. Core Components
The search engine can be conceptualized as having four primary components,
each with a distinct role in the discovery process.
1. User Interface (Conceptual): While not part of the search engine's
internal logic, a user interface is the entry point where a human user
or an orchestrating LLM provides the initial domain or topic for the
MCP server search. This interface could be a simple text prompt within
a chat application or a more structured input form.
2. LLM Orchestrator: This component is the "brain" of the search engine.
It is an LLM instance specifically tasked with interpreting the user's
query, extracting key search parameters, formulating appropriate search
strategies, and processing the results returned by various catalogs. The
LLM Orchestrator is responsible for understanding the semantic meaning
of the user's request and translating it into concrete actions.
3. Search Adapters/Connectors: These are specialized modules designed to
interface with different MCP server catalogs, directories, or libraries.
Each adapter understands the specific API, data format, and search
mechanisms of a particular catalog. This abstraction layer is crucial
for handling the heterogeneity of external information sources.
4. MCP Server Registry/Cache (Optional but Recommended): To improve
performance and reduce redundant calls to external catalogs, a local
registry or cache can store metadata about previously discovered MCP
servers. This component can also maintain an index of server
capabilities and domains, allowing for faster lookups and potentially
semantic search capabilities over known servers.
5. Result Presenter: This component takes the raw search results,
normalizes them, and presents them to the user or the requesting LLM
client in a clear, concise, and actionable format. It might involve
summarizing server functionalities, providing access endpoints, and
highlighting key features.
B. Data Flow
The process of searching for an MCP server through this engine follows a
logical sequence of steps.
1. A user or an LLM client submits a query specifying a domain or topic
(e.g., "I need an MCP server for real-time financial market data").
2. The query is received by the LLM Orchestrator, which analyzes the
natural language input to understand the user's intent, identify
keywords, and extract relevant search criteria (e.g., "financial",
"real-time", "market data").
3. Based on the extracted criteria, the LLM Orchestrator decides which
Search Adapters to invoke. It might generate specific queries tailored
to each catalog's API (e.g., a keyword search for one catalog, a
category filter for another).
4. The Search Adapters then communicate with their respective external MCP
server catalogs, submitting the formulated queries. These catalogs
respond with lists of MCP servers that match the search criteria, often
including metadata such as server name, description, and API endpoint.
5. The results from all invoked Search Adapters are collected by the LLM
Orchestrator. The orchestrator then processes these raw results, which
might involve deduplication, normalization of data formats, and further
semantic analysis to rank servers by relevance.
6. Finally, the Result Presenter formats the refined list of relevant MCP
servers into a user-friendly output, which is then returned to the
original user or LLM client. This output typically includes enough
information for the client to decide which server to interact with.
III. Implementing the LLM Orchestrator
The LLM Orchestrator is the heart of our search engine, responsible for
intelligent query processing and result management. Its implementation
involves leveraging the capabilities of a large language model for natural
language understanding and generation.
A. User Intent Parsing
The first critical step for the LLM Orchestrator is to accurately parse the
user's intent from their natural language query. This involves identifying
the core domain, topic, and any specific requirements for the MCP server.
Prompt engineering plays a crucial role here, guiding the LLM to extract
structured information from unstructured text.
Consider a user prompt like: "I need an MCP server that provides real-time
stock prices for the energy sector." The LLM Orchestrator would need to
extract:
- Domain: "financial data"
- Specificity: "real-time stock prices"
- Sector: "energy"
A simple prompt structure for the LLM could be:
"Analyze the following user request for an MCP server and extract the
primary domain, key features, and any specific sectors or data types.
Return the information as a JSON-like structure with 'domain', 'features',
and 'sector' keys. If a category is not explicitly mentioned, infer it
or leave it as 'general'.
User Request: {user_query}
import json
class LLMOrchestrator:
"""
The LLMOrchestrator is responsible for processing user queries,
interacting with the underlying LLM for intent parsing and query
formulation, and managing the overall search workflow.
"""
def __init__(self, llm_api_client):
"""
Initializes the LLMOrchestrator with an LLM API client.
The llm_api_client is expected to have a 'generate_response' method
that takes a prompt and returns a text response.
"""
self.llm_api_client = llm_api_client
def parse_user_intent(self, user_query: str) -> dict:
"""
Parses the user's natural language query to extract structured intent
using the underlying LLM.
Args:
user_query (str): The natural language query from the user.
Returns:
dict: A dictionary containing the extracted domain, features, and sector.
Example: {"domain": "financial data", "features": ["real-time stock prices"], "sector": "energy"}
"""
prompt = f"""
Analyze the following user request for an MCP server and extract the
primary domain, key features, and any specific sectors or data types.
Return the information as a JSON object with 'domain', 'features' (list of strings),
and 'sector' keys. If a category is not explicitly mentioned, infer it
or use 'general' for domain/sector and an empty list for features.
User Request: "{user_query}"
JSON Output:
"""
try:
llm_response = self.llm_api_client.generate_response(prompt)
# Attempt to parse the LLM's response as JSON
parsed_intent = json.loads(llm_response.strip())
return parsed_intent
except json.JSONDecodeError as e:
print(f"Error decoding LLM response JSON: {e}")
print(f"LLM Response: {self.llm_api_client.generate_response(prompt)}")
# Fallback to a default or error handling mechanism
return {"domain": "general", "features": [], "sector": "general"}
except Exception as e:
print(f"An unexpected error occurred during intent parsing: {e}")
return {"domain": "general", "features": [], "sector": "general"}
The `parse_user_intent` method demonstrates how the LLM Orchestrator interacts
with an underlying LLM API. The prompt is carefully crafted to instruct the
LLM to return a structured JSON output, making it easier for the orchestrator
to programmatically extract the necessary information. This structured output
is crucial for subsequent steps in the search process.
B. Query Formulation
Once the user's intent is parsed into a structured format, the LLM
Orchestrator needs to formulate specific queries for each of the available
Search Adapters. Different catalogs might have different query syntaxes or
expect different parameters. The orchestrator, potentially with further LLM
assistance or rule-based logic, translates the generic intent into
adapter-specific search requests.
For instance, if one Search Adapter connects to a public MCP server catalog
that supports keyword searches and category filters, the orchestrator would
construct a query suitable for that adapter. If another adapter connects to
an internal corporate catalog that uses product areas and specific asset types,
the orchestrator would generate a different query.
class LLMOrchestrator:
# ... (previous __init__ and parse_user_intent methods)
def formulate_adapter_queries(self, parsed_intent: dict, available_adapters: list) -> dict:
"""
Formulates adapter-specific queries based on the parsed user intent.
This method would typically use a combination of LLM capabilities
and predefined rules for each adapter type.
Args:
parsed_intent (dict): The structured intent extracted from the user query.
available_adapters (list): A list of SearchAdapter instances.
Returns:
dict: A dictionary where keys are adapter names and values are
the formulated queries for that adapter.
"""
adapter_queries = {}
domain = parsed_intent.get("domain", "general")
features = parsed_intent.get("features", [])
sector = parsed_intent.get("sector", "general")
# Example of how an LLM might help formulate queries for various adapters
# In a real system, this might involve more sophisticated prompt engineering
# or a mapping layer that understands each adapter's capabilities.
for adapter in available_adapters:
# This is a simplified example. A more advanced system might use the LLM
# to generate the *best* query string for each adapter based on its known schema.
if "publiccatalogadapter" in adapter.name.lower():
# For a generic public catalog, combine domain, features, and sector into a keyword query
query_string = f"{domain} {' '.join(features)} {sector} MCP server"
adapter_queries[adapter.name] = {"keywords": query_string}
elif "corporateassetsadapter" in adapter.name.lower():
# For a corporate asset catalog, map to specific parameters
asset_type = "ALL" # Could be refined by LLM if intent is very specific (e.g., "video")
collection = "ALL" # Could be refined by LLM
corporate_query = f"{domain} {' '.join(features)} {sector}"
adapter_queries[adapter.name] = {
"query": corporate_query,
"assetType": asset_type,
"collection": collection
}
elif "developerdocsadapter" in adapter.name.lower():
# For developer documentation, focus on API/SDK related terms
dev_query = f"{domain} API {' '.join(features)} {sector}"
adapter_queries[adapter.name] = {"query": dev_query, "includeContent": "true"}
else:
# Default keyword search for unknown adapters
adapter_queries[adapter.name] = {"keywords": f"{domain} {' '.join(features)} {sector}"}
return adapter_queries
The `formulate_adapter_queries` method takes the parsed intent and a list of
available Search Adapters. It then iterates through these adapters,
generating a tailored query for each. This example shows a rule-based approach
for simplicity, but in a more sophisticated system, the LLM itself could be
tasked with generating the optimal query parameters for each adapter, given
its understanding of the adapter's capabilities (which could be learned from
its documentation or schema).
C. Result Aggregation and Ranking
After the Search Adapters have executed their queries and returned results,
the LLM Orchestrator is responsible for aggregating these diverse results
and presenting them in a coherent and ranked manner. This often involves
normalizing data formats, deduplicating entries, and applying relevance
scoring. The LLM can play a significant role in semantic ranking,
understanding which results are most pertinent to the user's original intent,
even if keywords don't perfectly match.
class LLMOrchestrator:
# ... (previous methods)
def aggregate_and_rank_results(self, raw_results: dict, user_query: str) -> list:
"""
Aggregates raw search results from various adapters, deduplicates,
normalizes, and ranks them based on relevance to the original user query.
The LLM can be used here for semantic ranking.
Args:
raw_results (dict): A dictionary where keys are adapter names and
values are lists of MCP server results from that adapter.
user_query (str): The original natural language query for context.
Returns:
list: A sorted list of dictionaries, each representing a relevant
MCP server, with normalized fields (e.g., 'name', 'description', 'endpoint').
"""
all_servers = []
unique_server_identifiers = set()
# Step 1: Normalize and deduplicate results
for adapter_name, results_list in raw_results.items():
for server_data in results_list:
# Assuming each adapter returns a dictionary with at least 'name' and 'description'
# and potentially an 'endpoint' or 'access_info'.
# We need to normalize these to a common format.
normalized_server = {
"source_adapter": adapter_name,
"name": server_data.get("name", "Unknown Server"),
"description": server_data.get("description", "No description provided."),
"endpoint": server_data.get("endpoint", server_data.get("url", "N/A")),
"metadata": server_data # Keep original metadata for deeper inspection if needed
}
# Create a unique identifier for deduplication (e.g., based on name and endpoint)
server_identifier = f"{normalized_server['name']}-{normalized_server['endpoint']}"
if server_identifier not in unique_server_identifiers:
all_servers.append(normalized_server)
unique_server_identifiers.add(server_identifier)
if not all_servers:
return []
# Step 2: Use LLM for semantic ranking
# This prompt asks the LLM to score each server based on its description
# and the original user query.
ranking_prompt = f"""
Rank the following MCP servers based on their relevance to the user's original query:
"{user_query}"
Provide a relevance score (0-100, where 100 is most relevant) and a brief
reason for each server. Return the output as a JSON array of objects,
each with 'server_name', 'score', and 'reason'.
MCP Servers:
"""
for i, server in enumerate(all_servers):
ranking_prompt += f"\n{i+1}. Name: {server['name']}\n Description: {server['description']}\n"
ranking_prompt += "\nJSON Output:"
try:
llm_ranking_response = self.llm_api_client.generate_response(ranking_prompt)
ranking_data = json.loads(llm_ranking_response.strip())
# Map scores back to the original server objects
scored_servers = []
for server in all_servers:
found_score = next((item for item in ranking_data if item.get('server_name') == server['name']), None)
if found_score:
server['relevance_score'] = found_score.get('score', 0)
server['relevance_reason'] = found_score.get('reason', 'N/A')
else:
server['relevance_score'] = 0 # Default score if LLM didn't rank it
server['relevance_reason'] = 'Not explicitly ranked by LLM.'
scored_servers.append(server)
# Sort by relevance score in descending order
sorted_servers = sorted(scored_servers, key=lambda x: x.get('relevance_score', 0), reverse=True)
return sorted_servers
except json.JSONDecodeError as e:
print(f"Error decoding LLM ranking response JSON: {e}")
print(f"LLM Ranking Response: {self.llm_api_client.generate_response(ranking_prompt)}")
# Fallback: if LLM ranking fails, sort by some default (e.g., alphabetically or by source)
return sorted(all_servers, key=lambda x: x.get('name', ''))
except Exception as e:
print(f"An unexpected error occurred during result aggregation/ranking: {e}")
return sorted(all_servers, key=lambda x: x.get('name', ''))
The `aggregate_and_rank_results` method first normalizes and deduplicates
the results received from various adapters. It then constructs a prompt for
the LLM to semantically rank these servers based on their descriptions and
the original user query. This allows for a more intelligent ranking than
simple keyword matching. The LLM's response, expected in JSON, is then
parsed and used to sort the final list of servers, providing the most
relevant ones at the top.
IV. Search Adapters/Connectors
Search Adapters are the bridge between our LLM Orchestrator and the diverse
world of MCP server catalogs. Each adapter is responsible for knowing how to
query a specific catalog, parse its responses, and return results in a
standardized format. This modular design ensures that new catalogs can be
integrated by simply developing a new adapter without modifying the core
orchestrator logic.
A. Adapter Interface
To ensure consistency and ease of integration, all Search Adapters should
adhere to a common interface. This interface defines the methods that the
LLM Orchestrator will call to interact with any adapter.
from abc import ABC, abstractmethod
class SearchAdapter(ABC):
"""
Abstract base class for all Search Adapters.
Defines the common interface for interacting with MCP server catalogs.
"""
def __init__(self, name: str):
self.name = name
@abstractmethod
def search(self, query_params: dict) -> list:
"""
Executes a search query against the specific MCP server catalog
this adapter is responsible for.
Args:
query_params (dict): A dictionary of parameters specific to
this adapter's catalog API.
Returns:
list: A list of dictionaries, where each dictionary represents
an MCP server and contains its normalized metadata.
Example: [{"name": "Server A", "description": "...", "endpoint": "..."}]
"""
pass
@abstractmethod
def get_capabilities(self) -> dict:
"""
Returns a dictionary describing the capabilities and expected
query parameters of this adapter. This can be used by the LLM
Orchestrator for more intelligent query formulation.
"""
pass
The `SearchAdapter` abstract base class mandates two methods: `search` and
`get_capabilities`. The `search` method is where the actual interaction with
the external catalog happens, while `get_capabilities` allows the orchestrator
to understand what kind of queries an adapter can handle.
B. Concrete Adapter Implementations
Let's consider concrete implementations for a generic public catalog and
a corporate-specific catalog, leveraging a hypothetical external API.
1. Generic Public Catalog Adapter
This adapter would simulate interaction with a hypothetical public catalog
of MCP servers. In a real-world scenario, this might involve web scraping
or interacting with a public API. For our running example, we will simulate
a simple in-memory catalog.
class PublicCatalogAdapter(SearchAdapter):
"""
A concrete SearchAdapter for a hypothetical public MCP server catalog.
Simulates an in-memory catalog for demonstration purposes.
"""
def __init__(self, name="PublicCatalogAdapter"):
super().__init__(name)
# Simulate a simple in-memory catalog of MCP servers
self._catalog_data = [
{
"name": "Global Stock Data API",
"description": "Provides real-time and historical stock prices for global markets.",
"endpoint": "https://api.globalstockdata.com/v1/stocks",
"tags": ["financial", "real-time", "stocks", "market data"]
},
{
"name": "Weather Forecast Service",
"description": "Offers current weather conditions and forecasts worldwide.",
"endpoint": "https://api.weatherforecast.com/v2/forecast",
"tags": ["weather", "environmental", "real-time"]
},
{
"name": "Energy Market Insights",
"description": "Provides analytics and real-time data for the energy sector.",
"endpoint": "https://api.energymarketinsights.com/data",
"tags": ["financial", "energy", "market data", "analytics"]
},
{
"name": "Currency Exchange Rates",
"description": "Offers real-time currency conversion rates.",
"endpoint": "https://api.currencyexchange.com/rates",
"tags": ["financial", "currency", "real-time"]
}
]
def search(self, query_params: dict) -> list:
"""
Searches the simulated public catalog based on keywords.
"""
keywords = query_params.get("keywords", "").lower().split()
if not keywords:
return []
results = []
for server in self._catalog_data:
server_text = f"{server['name'].lower()} {server['description'].lower()} {' '.join(server['tags'])}"
if any(keyword in server_text for keyword in keywords):
results.append(server)
return results
def get_capabilities(self) -> dict:
"""
Describes the search capabilities of this adapter.
"""
return {
"search_type": "keyword_search",
"parameters": {"keywords": "string - space-separated keywords for searching"}
}
The `PublicCatalogAdapter` implements the `search` method by performing a
simple keyword match against its internal `_catalog_data`. This simulates
how an adapter would query an external service and filter results.
2. Corporate Assets Adapter
This adapter will leverage a hypothetical `external_api_client` to search
for corporate digital assets, which can be considered a type of MCP server
for content.
# Assume an external_api_client object is available in the environment
# from the tool context, or mocked for the running example.
class CorporateAssetsAdapter(SearchAdapter):
"""
A concrete SearchAdapter for corporate digital assets, utilizing a
hypothetical external API client.
"""
def __init__(self, name="CorporateAssetsAdapter", external_api_client=None):
super().__init__(name)
self.external_api_client = external_api_client # This will be the mock client in the example
def search(self, query_params: dict) -> list:
"""
Executes a search against corporate digital assets.
Expected query_params: {'query': str, 'assetType': str, 'collection': str}
"""
if not self.external_api_client:
print("External API client not initialized for CorporateAssetsAdapter.")
return []
query = query_params.get("query")
asset_type = query_params.get("assetType", "ALL")
collection = query_params.get("collection", "ALL")
max_results = query_params.get("maxResults", "10") # Default max results
if not query:
return []
try:
# This calls the mocked external tool function
response = self.external_api_client.get_corporate_assets(
query=query,
assetType=asset_type,
collection=collection,
maxResults=max_results
)
# Normalize the response for consistency
normalized_results = []
if response and "assets" in response:
for asset in response["assets"]:
normalized_results.append({
"name": asset.get("title", "Corporate Asset"),
"description": asset.get("description", "Corporate digital asset."),
"endpoint": asset.get("primaryDownloadUrl", "N/A"), # Or a more specific asset URL
"asset_id": asset.get("id"),
"asset_type": asset.get("assetType"),
"tags": asset.get("tags", [])
})
return normalized_results
except Exception as e:
print(f"Error calling Corporate Assets API: {e}")
return []
def get_capabilities(self) -> dict:
"""
Describes the search capabilities of this adapter for Corporate Assets.
"""
return {
"search_type": "corporate_assets_search",
"parameters": {
"query": "string - keyword search for assets",
"assetType": "string - IMAGES, VIDEOS, DOCUMENTS, AUDIO, ICONS, ALL",
"collection": "string - GENERAL, MARKETING, INTERNAL, ALL",
"maxResults": "string - Maximum number of results (5-50)"
}
}
The `CorporateAssetsAdapter` demonstrates how to integrate with a specific
external tool. It translates the generic `query_params` into the arguments
expected by the `get_corporate_assets` function and then normalizes
the tool's response into the standard server metadata format.
V. MCP Server Registry/Cache
For performance and efficiency, especially in scenarios with frequent
queries or slow external catalog APIs, an MCP Server Registry or Cache is
highly beneficial. This component acts as a local store for metadata about
previously discovered MCP servers, reducing the need to repeatedly query
external sources.
A. Purpose and Benefits
1. Reduced Latency: By serving results from a local cache, the search
engine can respond much faster to queries for commonly requested server
types.
2. Reduced API Calls: This minimizes the number of calls to external catalog
APIs, which can be rate-limited or incur costs.
3. Offline Capability: In some cases, a limited search capability might
even be possible if external catalogs are temporarily unavailable.
4. Enhanced Search: The local registry can be indexed and enriched with
additional metadata, allowing for more sophisticated internal search
capabilities (e.g., semantic search over cached descriptions).
B. Implementation Considerations
The registry can be implemented using a simple in-memory dictionary for small
scale, a persistent database (like SQLite, PostgreSQL) for larger deployments,
or a dedicated caching system (like Redis). Key design choices include:
- Data Structure: How to store server metadata (e.g., JSON documents,
relational tables).
- Indexing: How to efficiently search the cached data (e.g., full-text
indexing, tag-based indexing).
- Invalidation Strategy: How to ensure cached data remains fresh and
reflects changes in external catalogs (e.g., time-to-live (TTL),
webhook-based updates).
For our running example, we will use a simple in-memory cache that can be
extended to a persistent store.
import time
class MCPServerRegistry:
"""
A simple in-memory registry/cache for MCP server metadata.
This can be extended to use a persistent database for production.
"""
def __init__(self, cache_ttl_seconds=3600):
self._cache = {} # Stores server_identifier -> {"server_data": {}, "timestamp": float}
self.cache_ttl_seconds = cache_ttl_seconds
def add_server(self, server_data: dict):
"""
Adds or updates an MCP server's metadata in the cache.
A unique identifier is generated for each server.
"""
# Assuming 'name' and 'endpoint' uniquely identify a server
server_identifier = f"{server_data.get('name')}-{server_data.get('endpoint')}"
if server_identifier not in self._cache or self._is_stale(server_identifier):
self._cache[server_identifier] = {
"server_data": server_data,
"timestamp": time.time()
}
return True
return False # Server already exists and is fresh
def get_server(self, server_identifier: str) -> dict or None:
"""
Retrieves a server's metadata from the cache by its identifier.
Returns None if not found or if the entry is stale.
"""
if server_identifier in self._cache and not self._is_stale(server_identifier):
return self._cache[server_identifier]["server_data"]
return None
def search_registry(self, keywords: list) -> list:
"""
Performs a simple keyword search within the cached server descriptions.
This is a basic example; a real registry might use full-text search.
"""
matching_servers = []
for identifier, entry in self._cache.items():
if not self._is_stale(identifier):
server = entry["server_data"]
server_text = f"{server.get('name', '').lower()} {server.get('description', '').lower()}"
if any(keyword.lower() in server_text for keyword in keywords):
matching_servers.append(server)
return matching_servers
def _is_stale(self, server_identifier: str) -> bool:
"""
Checks if a cached entry is stale based on its timestamp and TTL.
"""
if server_identifier not in self._cache:
return True
return (time.time() - self._cache[server_identifier]["timestamp"]) > self.cache_ttl_seconds
def clear_stale_entries(self):
"""
Removes all stale entries from the cache.
"""
stale_identifiers = [
identifier for identifier in self._cache
if self._is_stale(identifier)
]
for identifier in stale_identifiers:
del self._cache[identifier]
print(f"Cleared {len(stale_identifiers)} stale entries from registry.")
The `MCPServerRegistry` provides basic `add_server`, `get_server`, and
`search_registry` functionalities. It includes a simple time-to-live (TTL)
mechanism to manage cache freshness. In a production environment, this
would be backed by a more robust and scalable data store.
VI. Result Presenter
The final component of our search engine is the Result Presenter. Its role
is to take the ranked list of MCP servers from the LLM Orchestrator and
format it into a human-readable or LLM-consumable output. The presentation
should be clear, concise, and provide enough information for the user or
client LLM to make an informed decision about which server to use.
A. Formatting for Readability
The presentation should include key details such as the server's name, a
brief description, its access endpoint, and the calculated relevance score.
Additional metadata, if relevant, can also be included.
class ResultPresenter:
"""
Formats and presents the aggregated and ranked MCP server search results.
"""
def present_results(self, ranked_servers: list) -> str:
"""
Formats the list of ranked MCP servers into a human-readable string.
Args:
ranked_servers (list): A list of dictionaries, each representing
a relevant MCP server, sorted by relevance.
Returns:
str: A formatted string presenting the search results.
"""
if not ranked_servers:
return "No MCP servers found matching your criteria."
output = "Found the following MCP servers, ranked by relevance:\n"
output += "---------------------------------------------------\n"
for i, server in enumerate(ranked_servers):
output += f"\n{i+1}. Server Name: {server.get('name', 'N/A')}\n"
output += f" Description: {server.get('description', 'N/A')}\n"
output += f" Endpoint: {server.get('endpoint', 'N/A')}\n"
output += f" Relevance Score: {server.get('relevance_score', 0)}/100\n"
output += f" Reason for Relevance: {server.get('relevance_reason', 'N/A')}\n"
output += f" Source Adapter: {server.get('source_adapter', 'N/A')}\n"
# Optionally include more metadata if desired
# output += f" Tags: {', '.join(server.get('metadata', {}).get('tags', []))}\n"
output += "---------------------------------------------------\n"
return output
def present_results_for_llm(self, ranked_servers: list) -> str:
"""
Formats the list of ranked MCP servers into a structured JSON string
suitable for consumption by another LLM client.
Args:
ranked_servers (list): A list of dictionaries, each representing
a relevant MCP server, sorted by relevance.
Returns:
str: A JSON string representing the search results.
"""
if not ranked_servers:
return json.dumps({"message": "No MCP servers found matching your criteria.", "servers": []})
# Create a simplified structure for LLM consumption
llm_output_servers = []
for server in ranked_servers:
llm_output_servers.append({
"name": server.get('name'),
"description": server.get('description'),
"endpoint": server.get('endpoint'),
"relevance_score": server.get('relevance_score')
})
return json.dumps({"servers": llm_output_servers})
The `ResultPresenter` offers two methods: `present_results` for human-readable
output and `present_results_for_llm` for structured JSON output, allowing
the search engine to serve both human users and other LLM clients effectively.
VII. Full Running Example: MCP Server Search Engine
This section provides a complete, runnable example integrating all the
components discussed. It simulates the LLM API client and the external API
client to make the example self-contained and executable.
A. Mock LLM API Client
To run the example without a live LLM API, we'll create a mock client that
simulates LLM responses based on predefined patterns.
class MockLLMAPIClient:
"""
A mock LLM API client for demonstration purposes.
Simulates responses for intent parsing and result ranking.
"""
def generate_response(self, prompt: str) -> str:
"""
Simulates an LLM response based on the prompt content.
"""
prompt_lower = prompt.lower()
if "analyze the following user request" in prompt_lower:
if "real-time stock prices for the energy sector" in prompt_lower:
return '{"domain": "financial data", "features": ["real-time stock prices"], "sector": "energy"}'
elif "industrial automation solutions" in prompt_lower:
return '{"domain": "industrial automation", "features": ["solutions"], "sector": "general"}'
elif "documents about plc programming" in prompt_lower:
return '{"domain": "documentation", "features": ["PLC programming"], "sector": "industrial"}'
elif "corporate financial api" in prompt_lower:
return '{"domain": "financial data", "features": ["API"], "sector": "general"}'
else:
return '{"domain": "general", "features": [], "sector": "general"}'
elif "rank the following mcp servers" in prompt_lower:
# Simple mock ranking logic based on keywords
ranking_output = []
server_lines = prompt.split("MCP Servers:")[1].split("\n")
server_data = []
current_server = {}
for line in server_lines:
line = line.strip()
if line.startswith(f"{len(server_data) + 1}. Name:"):
if current_server:
server_data.append(current_server)
current_server = {"server_name": line.replace(f"{len(server_data) + 1}. Name:", "").strip()}
elif line.startswith("Description:") and current_server:
current_server["description"] = line.replace("Description:", "").strip()
if current_server:
server_data.append(current_server)
for server in server_data:
score = 0
reason = "General relevance."
if "energy" in prompt_lower and "energy" in server.get("description", "").lower():
score += 30
reason = "Matches energy sector."
if "stock" in prompt_lower and "stock" in server.get("description", "").lower():
score += 40
reason = "Matches stock data."
if "real-time" in prompt_lower and "real-time" in server.get("description", "").lower():
score += 20
reason = "Matches real-time feature."
if "corporate" in prompt_lower and "corporate" in server.get("description", "").lower():
score += 50
reason = "Specifically requested corporate content."
if "plc programming" in prompt_lower and "plc" in server.get("description", "").lower():
score += 60
reason = "Matches PLC programming."
if "api" in prompt_lower and "api" in server.get("description", "").lower():
score += 35
reason = "Matches API feature."
score = min(100, score) # Cap score at 100
ranking_output.append({"server_name": server["server_name"], "score": score, "reason": reason})
return json.dumps(ranking_output)
return ""
B. Mock External API Client
Similarly, we'll create a mock for the external API client to simulate its
responses.
class MockExternalAPIClient:
"""
A mock client for external API tools, specifically for corporate assets and developer docs.
"""
def get_corporate_assets(self, query: str, assetType: str = "ALL", collection: str = "ALL", maxResults: str = "10") -> dict:
query_lower = query.lower()
mock_assets = []
if "industrial automation solutions" in query_lower:
mock_assets.append({
"id": "corp-ia-solution-1",
"title": "Advanced Automation System Series 7000",
"description": "Comprehensive automation solution for various industrial applications.",
"assetType": "DOCUMENT",
"primaryDownloadUrl": "https://corporate.com/docs/automation-series7000.pdf",
"tags": ["automation", "industrial", "control", "solution"]
})
mock_assets.append({
"id": "corp-ia-solution-2",
"title": "Edge Computing Devices Overview for Industry",
"description": "Edge computing solutions for industrial environments.",
"assetType": "VIDEO",
"primaryDownloadUrl": "https://corporate.com/videos/industrial-edge.mp4",
"tags": ["industrial", "edge", "computing", "solution"]
})
elif "plc programming" in query_lower:
mock_assets.append({
"id": "corp-plc-guide-1",
"title": "Unified Portal Programming Guide for Logic Controllers",
"description": "Detailed guide for programming industrial logic controllers using Unified Portal.",
"assetType": "DOCUMENT",
"primaryDownloadUrl": "https://corporate.com/docs/unified-portal-programming.pdf",
"tags": ["PLC", "programming", "logic controller"]
})
elif "energy" in query_lower:
mock_assets.append({
"id": "corp-energy-asset-1",
"title": "Corporate Energy Management Solutions Brochure",
"description": "Solutions for efficient energy consumption and distribution.",
"assetType": "DOCUMENT",
"primaryDownloadUrl": "https://corporate.com/docs/energy-management.pdf",
"tags": ["energy", "management", "solution"]
})
elif "financial" in query_lower and "api" in query_lower:
mock_assets.append({
"id": "corp-finance-api-doc-1",
"title": "Corporate Financial Services API Documentation",
"description": "Documentation for integrating with Corporate Financial Services APIs.",
"assetType": "DOCUMENT",
"primaryDownloadUrl": "https://corporate.com/docs/cfs-api.pdf",
"tags": ["financial", "API", "services"]
})
return {"assets": mock_assets[:int(maxResults)]}
def get_developer_docs(self, query: str, includeContent: str = "false", limit: str = "10") -> dict:
query_lower = query.lower()
mock_dev_docs = []
if "api" in query_lower and "financial" in query_lower:
mock_dev_docs.append({
"title": "Financial Data API Integration Guide",
"description": "Guide on integrating corporate financial data APIs.",
"url": "https://developer.corporate.com/docs/finance-api-guide",
"content_snippet": "This guide covers authentication and data retrieval for our financial APIs.",
"type": "API_DOC"
})
elif "industrial automation" in query_lower or "plc" in query_lower:
mock_dev_docs.append({
"title": "Industrial Edge API Reference",
"description": "API reference for developing applications on industrial edge platforms.",
"url": "https://developer.corporate.com/docs/industrial-edge-api",
"content_snippet": "Explore the APIs for device management and data processing on the edge.",
"type": "API_REFERENCE"
})
mock_dev_docs.append({
"title": "Logic Controller SDK Documentation",
"description": "Software Development Kit for industrial logic controllers.",
"url": "https://developer.corporate.com/docs/logic-controller-sdk",
"content_snippet": "Develop custom applications interacting with industrial PLCs.",
"type": "SDK_DOC"
})
return {"results": mock_dev_docs[:int(limit)]}
C. Main Application Logic
This is the main script that orchestrates the entire search process using
the components defined above.
# Import all necessary classes
# For a self-contained running example, we'll redefine the classes here
# to avoid circular imports or complex module structures for this single file.
import json
import time
from abc import ABC, abstractmethod
# --- Mock LLM API Client ---
class MockLLMAPIClient:
"""
A mock LLM API client for demonstration purposes.
Simulates responses for intent parsing and result ranking.
"""
def generate_response(self, prompt: str) -> str:
"""
Simulates an LLM response based on the prompt content.
"""
prompt_lower = prompt.lower()
if "analyze the following user request" in prompt_lower:
if "real-time stock prices for the energy sector" in prompt_lower:
return '{"domain": "financial data", "features": ["real-time stock prices"], "sector": "energy"}'
elif "industrial automation solutions" in prompt_lower:
return '{"domain": "industrial automation", "features": ["solutions"], "sector": "general"}'
elif "documents about plc programming" in prompt_lower:
return '{"domain": "documentation", "features": ["PLC programming"], "sector": "industrial"}'
elif "corporate financial api" in prompt_lower:
return '{"domain": "financial data", "features": ["API"], "sector": "general"}'
else:
return '{"domain": "general", "features": [], "sector": "general"}'
elif "rank the following mcp servers" in prompt_lower:
ranking_output = []
server_lines = prompt.split("MCP Servers:")[1].split("\n")
server_data = []
current_server = {}
for line in server_lines:
line = line.strip()
if line.startswith(f"{len(server_data) + 1}. Name:"):
if current_server:
server_data.append(current_server)
current_server = {"server_name": line.replace(f"{len(server_data) + 1}. Name:", "").strip()}
elif line.startswith("Description:") and current_server:
current_server["description"] = line.replace("Description:", "").strip()
if current_server:
server_data.append(current_server)
for server in server_data:
score = 0
reason = "General relevance."
if "energy" in prompt_lower and "energy" in server.get("description", "").lower():
score += 30
reason += " Matches energy sector."
if "stock" in prompt_lower and "stock" in server.get("description", "").lower():
score += 40
reason += " Matches stock data."
if "real-time" in prompt_lower and "real-time" in server.get("description", "").lower():
score += 20
reason += " Matches real-time feature."
if "corporate" in prompt_lower and "corporate" in server.get("description", "").lower():
score += 50
reason += " Specifically requested corporate content."
if "plc programming" in prompt_lower and "plc" in server.get("description", "").lower():
score += 60
reason += " Matches PLC programming."
if "api" in prompt_lower and "api" in server.get("description", "").lower():
score += 35
reason += " Matches API feature."
score = min(100, score)
ranking_output.append({"server_name": server["server_name"], "score": score, "reason": reason.strip()})
return json.dumps(ranking_output)
return ""
# --- Mock External API Client ---
class MockExternalAPIClient:
"""
A mock client for external API tools, specifically for corporate assets and developer docs.
"""
def get_corporate_assets(self, query: str, assetType: str = "ALL", collection: str = "ALL", maxResults: str = "10") -> dict:
query_lower = query.lower()
mock_assets = []
if "industrial automation solutions" in query_lower:
mock_assets.append({
"id": "corp-ia-solution-1",
"title": "Advanced Automation System Series 7000",
"description": "Comprehensive automation solution for various industrial applications.",
"assetType": "DOCUMENT",
"primaryDownloadUrl": "https://corporate.com/docs/automation-series7000.pdf",
"tags": ["automation", "industrial", "control", "solution"]
})
mock_assets.append({
"id": "corp-ia-solution-2",
"title": "Edge Computing Devices Overview for Industry",
"description": "Edge computing solutions for industrial environments.",
"assetType": "VIDEO",
"primaryDownloadUrl": "https://corporate.com/videos/industrial-edge.mp4",
"tags": ["industrial", "edge", "computing", "solution"]
})
elif "plc programming" in query_lower:
mock_assets.append({
"id": "corp-plc-guide-1",
"title": "Unified Portal Programming Guide for Logic Controllers",
"description": "Detailed guide for programming industrial logic controllers using Unified Portal.",
"assetType": "DOCUMENT",
"primaryDownloadUrl": "https://corporate.com/docs/unified-portal-programming.pdf",
"tags": ["PLC", "programming", "logic controller"]
})
elif "energy" in query_lower:
mock_assets.append({
"id": "corp-energy-asset-1",
"title": "Corporate Energy Management Solutions Brochure",
"description": "Solutions for efficient energy consumption and distribution.",
"assetType": "DOCUMENT",
"primaryDownloadUrl": "https://corporate.com/docs/energy-management.pdf",
"tags": ["energy", "management", "solution"]
})
elif "financial" in query_lower and "api" in query_lower:
mock_assets.append({
"id": "corp-finance-api-doc-1",
"title": "Corporate Financial Services API Documentation",
"description": "Documentation for integrating with Corporate Financial Services APIs.",
"assetType": "DOCUMENT",
"primaryDownloadUrl": "https://corporate.com/docs/cfs-api.pdf",
"tags": ["financial", "API", "services"]
})
return {"assets": mock_assets[:int(maxResults)]}
def get_developer_docs(self, query: str, includeContent: str = "false", limit: str = "10") -> dict:
query_lower = query.lower()
mock_dev_docs = []
if "api" in query_lower and "financial" in query_lower:
mock_dev_docs.append({
"title": "Financial Data API Integration Guide",
"description": "Guide on integrating corporate financial data APIs.",
"url": "https://developer.corporate.com/docs/finance-api-guide",
"content_snippet": "This guide covers authentication and data retrieval for our financial APIs.",
"type": "API_DOC"
})
elif "industrial automation" in query_lower or "plc" in query_lower:
mock_dev_docs.append({
"title": "Industrial Edge API Reference",
"description": "API reference for developing applications on industrial edge platforms.",
"url": "https://developer.corporate.com/docs/industrial-edge-api",
"content_snippet": "Explore the APIs for device management and data processing on the edge.",
"type": "API_REFERENCE"
})
mock_dev_docs.append({
"title": "Logic Controller SDK Documentation",
"description": "Software Development Kit for industrial logic controllers.",
"url": "https://developer.corporate.com/docs/logic-controller-sdk",
"content_snippet": "Develop custom applications interacting with industrial PLCs.",
"type": "SDK_DOC"
})
return {"results": mock_dev_docs[:int(limit)]}
# --- LLMOrchestrator Class ---
class LLMOrchestrator:
def __init__(self, llm_api_client):
self.llm_api_client = llm_api_client
def parse_user_intent(self, user_query: str) -> dict:
prompt = f"""
Analyze the following user request for an MCP server and extract the
primary domain, key features, and any specific sectors or data types.
Return the information as a JSON object with 'domain', 'features' (list of strings),
and 'sector' keys. If a category is not explicitly mentioned, infer it
or use 'general' for domain/sector and an empty list for features.
User Request: "{user_query}"
JSON Output:
"""
try:
llm_response = self.llm_api_client.generate_response(prompt)
parsed_intent = json.loads(llm_response.strip())
return parsed_intent
except json.JSONDecodeError as e:
print(f"Error decoding LLM response JSON: {e}")
return {"domain": "general", "features": [], "sector": "general"}
except Exception as e:
print(f"An unexpected error occurred during intent parsing: {e}")
return {"domain": "general", "features": [], "sector": "general"}
def formulate_adapter_queries(self, parsed_intent: dict, available_adapters: list) -> dict:
adapter_queries = {}
domain = parsed_intent.get("domain", "general")
features = parsed_intent.get("features", [])
sector = parsed_intent.get("sector", "general")
for adapter in available_adapters:
if "publiccatalogadapter" in adapter.name.lower():
query_string = f"{domain} {' '.join(features)} {sector} MCP server"
adapter_queries[adapter.name] = {"keywords": query_string}
elif "corporateassetsadapter" in adapter.name.lower():
asset_type = "ALL"
collection = "ALL"
corporate_query = f"{domain} {' '.join(features)} {sector}"
adapter_queries[adapter.name] = {
"query": corporate_query,
"assetType": asset_type,
"collection": collection
}
elif "developerdocsadapter" in adapter.name.lower():
dev_query = f"{domain} API {' '.join(features)} {sector}"
adapter_queries[adapter.name] = {"query": dev_query, "includeContent": "true"}
else:
adapter_queries[adapter.name] = {"keywords": f"{domain} {' '.join(features)} {sector}"}
return adapter_queries
def aggregate_and_rank_results(self, raw_results: dict, user_query: str) -> list:
all_servers = []
unique_server_identifiers = set()
for adapter_name, results_list in raw_results.items():
for server_data in results_list:
normalized_server = {
"source_adapter": adapter_name,
"name": server_data.get("name", server_data.get("title", "Unknown Server")),
"description": server_data.get("description", "No description provided."),
"endpoint": server_data.get("endpoint", server_data.get("primaryDownloadUrl", server_data.get("url", "N/A"))),
"metadata": server_data
}
server_identifier = f"{normalized_server['name']}-{normalized_server['endpoint']}"
if server_identifier not in unique_server_identifiers:
all_servers.append(normalized_server)
unique_server_identifiers.add(server_identifier)
if not all_servers:
return []
ranking_prompt = f"""
Rank the following MCP servers based on their relevance to the user's original query:
"{user_query}"
Provide a relevance score (0-100, where 100 is most relevant) and a brief
reason for each server. Return the output as a JSON array of objects,
each with 'server_name', 'score', and 'reason'.
MCP Servers:
"""
for i, server in enumerate(all_servers):
ranking_prompt += f"\n{i+1}. Name: {server['name']}\n Description: {server['description']}\n"
ranking_prompt += "\nJSON Output:"
try:
llm_ranking_response = self.llm_api_client.generate_response(ranking_prompt)
ranking_data = json.loads(llm_ranking_response.strip())
scored_servers = []
for server in all_servers:
found_score = next((item for item in ranking_data if item.get('server_name') == server['name']), None)
if found_score:
server['relevance_score'] = found_score.get('score', 0)
server['relevance_reason'] = found_score.get('reason', 'N/A')
else:
server['relevance_score'] = 0
server['relevance_reason'] = 'Not explicitly ranked by LLM.'
scored_servers.append(server)
sorted_servers = sorted(scored_servers, key=lambda x: x.get('relevance_score', 0), reverse=True)
return sorted_servers
except json.JSONDecodeError as e:
print(f"Error decoding LLM ranking response JSON: {e}")
return sorted(all_servers, key=lambda x: x.get('name', ''))
except Exception as e:
print(f"An unexpected error occurred during result aggregation/ranking: {e}")
return sorted(all_servers, key=lambda x: x.get('name', ''))
# --- SearchAdapter Abstract Base Class ---
class SearchAdapter(ABC):
def __init__(self, name: str):
self.name = name
@abstractmethod
def search(self, query_params: dict) -> list:
pass
@abstractmethod
def get_capabilities(self) -> dict:
pass
# --- PublicCatalogAdapter ---
class PublicCatalogAdapter(SearchAdapter):
def __init__(self, name="PublicCatalogAdapter"):
super().__init__(name)
self._catalog_data = [
{
"name": "Global Stock Data API",
"description": "Provides real-time and historical stock prices for global markets.",
"endpoint": "https://api.globalstockdata.com/v1/stocks",
"tags": ["financial", "real-time", "stocks", "market data"]
},
{
"name": "Weather Forecast Service",
"description": "Offers current weather conditions and forecasts worldwide.",
"endpoint": "https://api.weatherforecast.com/v2/forecast",
"tags": ["weather", "environmental", "real-time"]
},
{
"name": "Energy Market Insights",
"description": "Provides analytics and real-time data for the energy sector.",
"endpoint": "https://api.energymarketinsights.com/data",
"tags": ["financial", "energy", "market data", "analytics"]
},
{
"name": "Currency Exchange Rates",
"description": "Offers real-time currency conversion rates.",
"endpoint": "https://api.currencyexchange.com/rates",
"tags": ["financial", "currency", "real-time"]
}
]
def search(self, query_params: dict) -> list:
keywords = query_params.get("keywords", "").lower().split()
if not keywords:
return []
results = []
for server in self._catalog_data:
server_text = f"{server['name'].lower()} {server['description'].lower()} {' '.join(server['tags'])}"
if any(keyword in server_text for keyword in keywords):
results.append(server)
return results
def get_capabilities(self) -> dict:
return {
"search_type": "keyword_search",
"parameters": {"keywords": "string - space-separated keywords for searching"}
}
# --- CorporateAssetsAdapter ---
class CorporateAssetsAdapter(SearchAdapter):
def __init__(self, name="CorporateAssetsAdapter", external_api_client=None):
super().__init__(name)
self.external_api_client = external_api_client
def search(self, query_params: dict) -> list:
if not self.external_api_client:
print("External API client not initialized for CorporateAssetsAdapter.")
return []
query = query_params.get("query")
asset_type = query_params.get("assetType", "ALL")
collection = query_params.get("collection", "ALL")
max_results = query_params.get("maxResults", "10")
if not query:
return []
try:
response = self.external_api_client.get_corporate_assets(
query=query,
assetType=asset_type,
collection=collection,
maxResults=max_results
)
normalized_results = []
if response and "assets" in response:
for asset in response["assets"]:
normalized_results.append({
"name": asset.get("title", "Corporate Asset"),
"description": asset.get("description", "Corporate digital asset."),
"endpoint": asset.get("primaryDownloadUrl", "N/A"),
"asset_id": asset.get("id"),
"asset_type": asset.get("assetType"),
"tags": asset.get("tags", [])
})
return normalized_results
except Exception as e:
print(f"Error calling Corporate Assets API: {e}")
return []
def get_capabilities(self) -> dict:
return {
"search_type": "corporate_assets_search",
"parameters": {
"query": "string - keyword search for assets",
"assetType": "string - IMAGES, VIDEOS, DOCUMENTS, AUDIO, ICONS, ALL",
"collection": "string - GENERAL, MARKETING, INTERNAL, ALL",
"maxResults": "string - Maximum number of results (5-50)"
}
}
# --- DeveloperDocsAdapter (New Adapter) ---
class DeveloperDocsAdapter(SearchAdapter):
"""
A concrete SearchAdapter for developer documentation, utilizing a
hypothetical external API client.
"""
def __init__(self, name="DeveloperDocsAdapter", external_api_client=None):
super().__init__(name)
self.external_api_client = external_api_client
def search(self, query_params: dict) -> list:
if not self.external_api_client:
print("External API client not initialized for DeveloperDocsAdapter.")
return []
query = query_params.get("query")
include_content = query_params.get("includeContent", "false")
limit = query_params.get("limit", "10")
if not query:
return []
try:
response = self.external_api_client.get_developer_docs(
query=query,
includeContent=include_content,
limit=limit
)
normalized_results = []
if response and "results" in response:
for doc in response["results"]:
normalized_results.append({
"name": doc.get("title", "Developer Doc"),
"description": doc.get("description", "Developer documentation."),
"endpoint": doc.get("url", "N/A"),
"doc_type": doc.get("type"),
"content_snippet": doc.get("content_snippet", "")
})
return normalized_results
except Exception as e:
print(f"Error calling Developer Docs API: {e}")
return []
def get_capabilities(self) -> dict:
return {
"search_type": "developer_docs_search",
"parameters": {
"query": "string - search query for developer documentation",
"includeContent": "string - 'true' or 'false' to include content snippets",
"limit": "string - maximum number of results"
}
}
# --- MCPServerRegistry ---
class MCPServerRegistry:
def __init__(self, cache_ttl_seconds=3600):
self._cache = {}
self.cache_ttl_seconds = cache_ttl_seconds
def add_server(self, server_data: dict):
server_identifier = f"{server_data.get('name')}-{server_data.get('endpoint')}"
if server_identifier not in self._cache or self._is_stale(server_identifier):
self._cache[server_identifier] = {
"server_data": server_data,
"timestamp": time.time()
}
return True
return False
def get_server(self, server_identifier: str) -> dict or None:
if server_identifier in self._cache and not self._is_stale(server_identifier):
return self._cache[server_identifier]["server_data"]
return None
def search_registry(self, keywords: list) -> list:
matching_servers = []
for identifier, entry in self._cache.items():
if not self._is_stale(identifier):
server = entry["server_data"]
server_text = f"{server.get('name', '').lower()} {server.get('description', '').lower()}"
if any(keyword.lower() in server_text for keyword in keywords):
matching_servers.append(server)
return matching_servers
def _is_stale(self, server_identifier: str) -> bool:
if server_identifier not in self._cache:
return True
return (time.time() - self._cache[server_identifier]["timestamp"]) > self.cache_ttl_seconds
def clear_stale_entries(self):
stale_identifiers = [
identifier for identifier in self._cache
if self._is_stale(identifier)
]
for identifier in stale_identifiers:
del self._cache[identifier]
print(f"Cleared {len(stale_identifiers)} stale entries from registry.")
# --- ResultPresenter ---
class ResultPresenter:
def present_results(self, ranked_servers: list) -> str:
if not ranked_servers:
return "No MCP servers found matching your criteria."
output = "Found the following MCP servers, ranked by relevance:\n"
output += "---------------------------------------------------\n"
for i, server in enumerate(ranked_servers):
output += f"\n{i+1}. Server Name: {server.get('name', 'N/A')}\n"
output += f" Description: {server.get('description', 'N/A')}\n"
output += f" Endpoint: {server.get('endpoint', 'N/A')}\n"
output += f" Relevance Score: {server.get('relevance_score', 0)}/100\n"
output += f" Reason for Relevance: {server.get('relevance_reason', 'N/A')}\n"
output += f" Source Adapter: {server.get('source_adapter', 'N/A')}\n"
output += "---------------------------------------------------\n"
return output
def present_results_for_llm(self, ranked_servers: list) -> str:
if not ranked_servers:
return json.dumps({"message": "No MCP servers found matching your criteria.", "servers": []})
llm_output_servers = []
for server in ranked_servers:
llm_output_servers.append({
"name": server.get('name'),
"description": server.get('description'),
"endpoint": server.get('endpoint'),
"relevance_score": server.get('relevance_score')
})
return json.dumps({"servers": llm_output_servers})
# --- Main Application Class ---
class MCPServerSearchEngine:
"""
The main class that orchestrates the MCP Server Search Engine.
"""
def __init__(self, llm_api_client, external_api_client):
self.llm_orchestrator = LLMOrchestrator(llm_api_client)
self.registry = MCPServerRegistry()
self.result_presenter = ResultPresenter()
# Initialize all available Search Adapters
self.adapters = [
PublicCatalogAdapter(),
CorporateAssetsAdapter(external_api_client=external_api_client),
DeveloperDocsAdapter(external_api_client=external_api_client)
# Add more adapters here as needed
]
def search_mcp_servers(self, user_query: str) -> str:
"""
Executes a full search for MCP servers based on the user's query.
Args:
user_query (str): The natural language query from the user.
Returns:
str: A formatted string of search results.
"""
print(f"\n--- Processing User Query: '{user_query}' ---")
# 1. Parse User Intent
print("Step 1: Parsing user intent...")
parsed_intent = self.llm_orchestrator.parse_user_intent(user_query)
print(f" Parsed Intent: {json.dumps(parsed_intent, indent=2)}")
# 2. Formulate Adapter Queries
print("Step 2: Formulating adapter-specific queries...")
adapter_queries = self.llm_orchestrator.formulate_adapter_queries(parsed_intent, self.adapters)
print(f" Adapter Queries: {json.dumps(adapter_queries, indent=2)}")
# 3. Execute Searches via Adapters and populate Registry
print("Step 3: Executing searches via adapters and populating registry...")
raw_results = {}
for adapter in self.adapters:
adapter_name = adapter.name
query_params = adapter_queries.get(adapter_name)
if query_params:
print(f" Searching with {adapter_name} using params: {query_params}")
adapter_results = adapter.search(query_params)
raw_results[adapter_name] = adapter_results
# Add results to registry
for server in adapter_results:
self.registry.add_server(server)
else:
print(f" No specific query formulated for {adapter_name}.")
# Also search the local registry for cached results
registry_keywords = []
if parsed_intent.get("domain"):
registry_keywords.append(parsed_intent["domain"])
registry_keywords.extend(parsed_intent.get("features", []))
if parsed_intent.get("sector"):
registry_keywords.append(parsed_intent["sector"])
if registry_keywords:
print(f" Searching local registry with keywords: {registry_keywords}")
cached_results = self.registry.search_registry(registry_keywords)
if "Registry" not in raw_results:
raw_results["Registry"] = []
raw_results["Registry"].extend(cached_results)
else:
print(" No keywords for registry search.")
# 4. Aggregate and Rank Results
print("Step 4: Aggregating and ranking results...")
ranked_servers = self.llm_orchestrator.aggregate_and_rank_results(raw_results, user_query)
# 5. Present Results
print("Step 5: Presenting results...")
final_output = self.result_presenter.present_results(ranked_servers)
return final_output
# --- Main Execution Block ---
if __name__ == "__main__":
# Initialize mock clients
mock_llm_client = MockLLMAPIClient()
mock_external_api_client = MockExternalAPIClient()
# Create the search engine instance
search_engine = MCPServerSearchEngine(mock_llm_client, mock_external_api_client)
# Example User Queries
queries = [
"I need an MCP server that provides real-time stock prices for the energy sector.",
"Find industrial automation solutions.",
"I'm looking for documents about PLC programming.",
"Show me any corporate financial APIs."
]
for query in queries:
results = search_engine.search_mcp_servers(query)
print(results)
print("\n" + "="*80 + "\n") # Separator for queries
# Demonstrate cache behavior (run a query again)
print("\n--- Rerunning a query to demonstrate registry caching ---")
results_cached = search_engine.search_mcp_servers(queries[0])
print(results_cached)
print("\n" + "="*80 + "\n")
# Clear stale entries (example of maintenance)
search_engine.registry.clear_stale_entries()
This comprehensive running example illustrates the full workflow of the
LLM-based MCP Server Search Engine. It starts by initializing mock clients
for the LLM and external APIs, then sets up the `MCPServerSearchEngine` with
its orchestrator, registry, and adapters. It processes several example user
queries, demonstrating how intent is parsed, queries are formulated for
different adapters, results are collected and ranked, and finally presented.
The example also includes a demonstration of the registry's caching behavior.
Conclusion
The implementation of an LLM-based MCP Server Search Engine represents a significant step towards unlocking the full potential of the Model Context
Protocol ecosystem. By leveraging the natural language understanding and
generation capabilities of large language models, this engine provides an
intelligent and adaptive mechanism for discovering relevant MCP servers.
The modular architecture, comprising an LLM Orchestrator, flexible Search
Adapters, an efficient MCP Server Registry, and a clear Result Presenter,
ensures scalability, maintainability, and extensibility. As the landscape
of MCP servers continues to evolve, such a search engine will be
indispensable for LLM clients to seamlessly integrate with and utilize the
vast array of external services and data sources, ultimately enhancing their
utility and responsiveness to complex user demands. This guide has provided
a deep dive into the conceptual and practical aspects of building such a
system, complete with a runnable example to illustrate its core functionalities.
