Introduction
The Model Context Protocol (MCP) developed by Anthropic has emerged as a revolutionary standard for connecting Large Language Models to external data sources and tools. As the MCP ecosystem grows rapidly with thousands of available servers, developers face the challenge of discovering the right MCP servers for their specific needs. This article presents a comprehensive tutorial for building an intelligent agent that leverages a local LLM to help developers find appropriate MCP servers through natural language queries.
Our agent will accept natural language descriptions of desired functionality and search both general web sources and specialized MCP directories to find relevant servers. For each discovered server, the agent will provide detailed summaries including functionality descriptions, access URLs, and implementation details.
Understanding the Problem Domain
The Model Context Protocol ecosystem currently includes over 3,900 servers spanning categories from data storage and content search to AI memory and development tools. These servers enable LLMs to interact with external systems including databases, APIs, file systems, and specialized services. However, finding the right server for a specific use case requires navigating multiple directories and understanding technical specifications.
Our agent addresses this challenge by providing an intelligent search interface that understands developer intent and returns curated results with comprehensive analysis.
Architecture Overview
The agent consists of several interconnected components working together to process natural language queries and return structured results. The core architecture includes a local LLM for natural language processing, web search capabilities for discovering MCP servers, result analysis and summarization logic, and a presentation layer for formatted output.
The local LLM serves as the central intelligence, interpreting user queries, generating search terms, analyzing discovered content, and producing human-readable summaries. The web search component queries both general search engines and specialized MCP directories to find relevant servers. The analysis engine processes raw search results to extract meaningful information about each server's capabilities and implementation details.
Setting Up the Local LLM Environment
We begin by establishing a local LLM environment using Ollama, which provides an excellent foundation for running quantized models locally. Ollama supports the GGUF format and offers a simple API for integration.
First, install Ollama following the official installation instructions for your platform. For this implementation, we recommend using a capable model such as Mistral 7B or Llama 2 7B, which provide good performance for text analysis and generation tasks while remaining accessible on consumer hardware.
import ollama
import json
import re
import requests
from typing import List, Dict, Any
from dataclasses import dataclass
from urllib.parse import quote_plus
import time
@dataclass
class MCPServerInfo:
"""Data structure for storing MCP server information"""
name: str
description: str
url: str
functionality: str
category: str
provider: str
installation_info: str
class LocalLLMClient:
"""Client for interacting with local LLM via Ollama"""
def __init__(self, model_name: str = "mistral:7b"):
self.model_name = model_name
self.client = ollama.Client()
def generate_response(self, prompt: str, system_message: str = "") -> str:
"""Generate response from local LLM"""
try:
messages = []
if system_message:
messages.append({"role": "system", "content": system_message})
messages.append({"role": "user", "content": prompt})
response = self.client.chat(
model=self.model_name,
messages=messages,
stream=False
)
return response['message']['content']
except Exception as e:
print(f"Error generating LLM response: {e}")
return ""
The LocalLLMClient class provides a clean interface for interacting with our local model. The generate_response method accepts both user prompts and system messages, allowing us to provide context and instructions for different tasks throughout the agent's operation.
Implementing Web Search Functionality
The web search component forms the foundation of our server discovery process. We implement multiple search strategies to ensure comprehensive coverage of available MCP servers.
class WebSearcher:
"""Handles web searches for MCP server discovery"""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
# Known MCP directories and catalogs
self.mcp_directories = [
"https://mcplist.ai",
"https://openmcp.directory",
"https://github.com/modelcontextprotocol/servers",
"https://mcp-catalog.com"
]
def search_general_web(self, query: str) -> List[Dict[str, Any]]:
"""Search general web for MCP servers using DuckDuckGo"""
search_url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
try:
response = self.session.get(search_url, timeout=10)
response.raise_for_status()
# Parse search results from DuckDuckGo HTML
results = self._parse_duckduckgo_results(response.text)
return results[:10] # Limit to top 10 results
except Exception as e:
print(f"Error in general web search: {e}")
return []
def search_mcp_directories(self, query: str) -> List[Dict[str, Any]]:
"""Search specialized MCP directories"""
all_results = []
for directory_url in self.mcp_directories:
try:
# Search within each MCP directory
results = self._search_directory(directory_url, query)
all_results.extend(results)
time.sleep(1) # Rate limiting
except Exception as e:
print(f"Error searching {directory_url}: {e}")
continue
return all_results
def _parse_duckduckgo_results(self, html_content: str) -> List[Dict[str, Any]]:
"""Parse DuckDuckGo search results from HTML"""
results = []
# Regular expressions to extract search results
title_pattern = r'<a[^>]*class="result__a"[^>]*href="([^"]*)"[^>]*>([^<]*)</a>'
snippet_pattern = r'<a[^>]*class="result__snippet"[^>]*>([^<]*)</a>'
titles = re.findall(title_pattern, html_content)
snippets = re.findall(snippet_pattern, html_content)
for i, (url, title) in enumerate(titles):
snippet = snippets[i] if i < len(snippets) else ""
results.append({
'title': title.strip(),
'url': url.strip(),
'snippet': snippet.strip()
})
return results
def _search_directory(self, directory_url: str, query: str) -> List[Dict[str, Any]]:
"""Search within a specific MCP directory"""
try:
# Fetch directory content
response = self.session.get(directory_url, timeout=10)
response.raise_for_status()
# Extract MCP server listings from directory
servers = self._extract_servers_from_directory(response.text, directory_url)
# Filter servers based on query relevance
relevant_servers = self._filter_relevant_servers(servers, query)
return relevant_servers
except Exception as e:
print(f"Error searching directory {directory_url}: {e}")
return []
def _extract_servers_from_directory(self, html_content: str, base_url: str) -> List[Dict[str, Any]]:
"""Extract MCP server information from directory HTML"""
servers = []
# Pattern to match server entries in directory listings
server_pattern = r'<div[^>]*class="[^"]*server[^"]*"[^>]*>(.*?)</div>'
name_pattern = r'<h[1-6][^>]*>([^<]+)</h[1-6]>'
desc_pattern = r'<p[^>]*class="[^"]*description[^"]*"[^>]*>([^<]+)</p>'
link_pattern = r'<a[^>]*href="([^"]*)"[^>]*>'
server_blocks = re.findall(server_pattern, html_content, re.DOTALL)
for block in server_blocks:
name_match = re.search(name_pattern, block)
desc_match = re.search(desc_pattern, block)
link_match = re.search(link_pattern, block)
if name_match:
server = {
'name': name_match.group(1).strip(),
'description': desc_match.group(1).strip() if desc_match else "",
'url': link_match.group(1).strip() if link_match else base_url,
'source': base_url
}
servers.append(server)
return servers
def _filter_relevant_servers(self, servers: List[Dict[str, Any]], query: str) -> List[Dict[str, Any]]:
"""Filter servers based on query relevance"""
query_terms = query.lower().split()
relevant_servers = []
for server in servers:
# Calculate relevance score based on term matches
text_to_search = f"{server.get('name', '')} {server.get('description', '')}".lower()
score = 0
for term in query_terms:
if term in text_to_search:
score += 1
# Include servers with at least one matching term
if score > 0:
server['relevance_score'] = score
relevant_servers.append(server)
# Sort by relevance score
relevant_servers.sort(key=lambda x: x['relevance_score'], reverse=True)
return relevant_servers
The WebSearcher class implements a comprehensive search strategy that covers both general web searches and specialized MCP directories. The search_general_web method uses DuckDuckGo to find MCP-related content across the internet, while search_mcp_directories focuses on known catalogs of MCP servers.
The implementation includes robust error handling and rate limiting to ensure reliable operation. The _filter_relevant_servers method provides basic relevance scoring to prioritize results that best match the user's query.
Creating the MCP Server Discovery Logic
The core discovery logic combines search results with intelligent analysis to identify and categorize MCP servers based on user requirements.
class MCPServerDiscovery:
"""Main class for discovering and analyzing MCP servers"""
def __init__(self, llm_client: LocalLLMClient):
self.llm_client = llm_client
self.web_searcher = WebSearcher()
def discover_servers(self, user_query: str) -> List[MCPServerInfo]:
"""Main method to discover MCP servers based on user query"""
# Step 1: Generate search terms from user query
search_terms = self._generate_search_terms(user_query)
# Step 2: Perform searches
all_results = []
for term in search_terms:
# Search general web
web_results = self.web_searcher.search_general_web(f"MCP server {term}")
all_results.extend(web_results)
# Search MCP directories
directory_results = self.web_searcher.search_mcp_directories(term)
all_results.extend(directory_results)
# Step 3: Deduplicate and filter results
unique_results = self._deduplicate_results(all_results)
# Step 4: Analyze and extract server information
server_infos = []
for result in unique_results[:20]: # Limit to top 20 for analysis
server_info = self._analyze_server_result(result, user_query)
if server_info:
server_infos.append(server_info)
# Step 5: Rank and return top results
ranked_servers = self._rank_servers(server_infos, user_query)
return ranked_servers[:10] # Return top 10 results
def _generate_search_terms(self, user_query: str) -> List[str]:
"""Generate relevant search terms from user query using LLM"""
system_message = """You are an expert at understanding developer needs for MCP (Model Context Protocol) servers.
Given a user query about what kind of MCP server they need, generate 3-5 specific search terms that would help find relevant servers.
Focus on:
- Technical keywords related to the functionality
- Data source types mentioned
- Integration types needed
- Specific tools or platforms mentioned
Return only the search terms, one per line, without explanations."""
prompt = f"User query: {user_query}\n\nGenerate search terms:"
response = self.llm_client.generate_response(prompt, system_message)
# Parse search terms from response
terms = [term.strip() for term in response.split('\n') if term.strip()]
# Add some default MCP-related terms
terms.extend(['model context protocol', 'MCP server'])
return list(set(terms)) # Remove duplicates
def _deduplicate_results(self, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Remove duplicate search results based on URL and title similarity"""
seen_urls = set()
unique_results = []
for result in results:
url = result.get('url', '')
title = result.get('title', result.get('name', ''))
# Create a normalized identifier
identifier = f"{url.lower()}|{title.lower()}"
if identifier not in seen_urls:
seen_urls.add(identifier)
unique_results.append(result)
return unique_results
def _analyze_server_result(self, result: Dict[str, Any], user_query: str) -> MCPServerInfo:
"""Analyze a search result to extract MCP server information"""
# Fetch additional content if needed
content = self._fetch_page_content(result.get('url', ''))
system_message = """You are an expert at analyzing MCP (Model Context Protocol) servers.
Given information about a potential MCP server, extract the following details:
1. Server name
2. Brief description of functionality
3. Main category (e.g., Data Storage, Content Search, AI Memory, etc.)
4. Provider/maintainer
5. Installation information
6. Specific capabilities and features
If this is not actually an MCP server, return "NOT_MCP_SERVER".
Format your response as JSON with these fields:
{
"name": "server name",
"description": "brief description",
"category": "main category",
"provider": "provider name",
"installation_info": "how to install/use",
"functionality": "detailed capabilities"
}"""
prompt = f"""
Search Result:
Title: {result.get('title', result.get('name', 'Unknown'))}
URL: {result.get('url', 'Unknown')}
Description: {result.get('snippet', result.get('description', 'No description'))}
Additional Content: {content[:2000]} # Limit content length
User Query Context: {user_query}
Analyze this information:
"""
response = self.llm_client.generate_response(prompt, system_message)
if "NOT_MCP_SERVER" in response:
return None
try:
# Parse JSON response
server_data = json.loads(response)
return MCPServerInfo(
name=server_data.get('name', 'Unknown'),
description=server_data.get('description', ''),
url=result.get('url', ''),
functionality=server_data.get('functionality', ''),
category=server_data.get('category', 'Unknown'),
provider=server_data.get('provider', 'Unknown'),
installation_info=server_data.get('installation_info', '')
)
except json.JSONDecodeError:
# Fallback to basic extraction if JSON parsing fails
return self._basic_server_extraction(result)
def _fetch_page_content(self, url: str) -> str:
"""Fetch and extract text content from a webpage"""
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
# Basic HTML text extraction
text = re.sub(r'<[^>]+>', ' ', response.text)
text = re.sub(r'\s+', ' ', text).strip()
return text[:5000] # Limit content length
except Exception as e:
print(f"Error fetching content from {url}: {e}")
return ""
def _basic_server_extraction(self, result: Dict[str, Any]) -> MCPServerInfo:
"""Basic fallback extraction when LLM analysis fails"""
return MCPServerInfo(
name=result.get('title', result.get('name', 'Unknown Server')),
description=result.get('snippet', result.get('description', 'No description available')),
url=result.get('url', ''),
functionality='Analysis not available',
category='Unknown',
provider='Unknown',
installation_info='See server documentation'
)
def _rank_servers(self, servers: List[MCPServerInfo], user_query: str) -> List[MCPServerInfo]:
"""Rank servers based on relevance to user query"""
system_message = """You are an expert at matching MCP servers to user requirements.
Given a user query and a list of MCP servers, rank them by relevance.
Consider:
- How well the server's functionality matches the user's needs
- The quality and completeness of the server
- The reputation of the provider
- How actively maintained the server appears to be
Return only the server names in order of relevance, one per line."""
server_summaries = []
for i, server in enumerate(servers):
summary = f"{i+1}. {server.name}: {server.description} (Category: {server.category})"
server_summaries.append(summary)
prompt = f"""
User Query: {user_query}
Available Servers:
{chr(10).join(server_summaries)}
Rank by relevance:
"""
response = self.llm_client.generate_response(prompt, system_message)
# Parse ranking and reorder servers
ranked_names = [name.strip() for name in response.split('\n') if name.strip()]
# Create ranking map
ranking_map = {}
for i, name in enumerate(ranked_names):
# Extract server name from numbered list
clean_name = re.sub(r'^\d+\.\s*', '', name)
ranking_map[clean_name] = i
# Sort servers based on ranking
def get_rank(server):
for ranked_name in ranking_map:
if ranked_name.lower() in server.name.lower():
return ranking_map[ranked_name]
return len(servers) # Put unranked servers at the end
return sorted(servers, key=get_rank)
The MCPServerDiscovery class orchestrates the entire discovery process. The discover_servers method implements a multi-step pipeline that transforms natural language queries into structured server information.
The _generate_search_terms method leverages the local LLM to understand user intent and generate relevant search terms. This approach ensures that searches capture both explicit requirements and implicit needs that might not be directly stated in the original query.
The _analyze_server_result method performs deep analysis of each discovered result, using the LLM to extract structured information about server capabilities, installation requirements, and provider details.
Building Natural Language Query Processing
The natural language processing component enables users to express their requirements in conversational terms while ensuring accurate interpretation of technical needs.
class QueryProcessor:
"""Processes and interprets natural language queries for MCP server discovery"""
def __init__(self, llm_client: LocalLLMClient):
self.llm_client = llm_client
def process_query(self, user_input: str) -> Dict[str, Any]:
"""Process user query and extract structured requirements"""
system_message = """You are an expert at understanding developer requirements for MCP (Model Context Protocol) servers.
Analyze the user's query and extract:
1. Primary functionality needed
2. Data sources or systems to integrate with
3. Specific technologies or platforms mentioned
4. Use case category (e.g., data access, tool integration, AI enhancement)
5. Any constraints or preferences
Return your analysis as JSON with these fields:
{
"primary_function": "main functionality needed",
"data_sources": ["list", "of", "data", "sources"],
"technologies": ["mentioned", "technologies"],
"category": "use case category",
"constraints": ["any", "constraints"],
"clarified_query": "expanded and clarified version of the query"
}"""
prompt = f"User query: {user_input}\n\nAnalyze this query:"
response = self.llm_client.generate_response(prompt, system_message)
try:
return json.loads(response)
except json.JSONDecodeError:
# Fallback to basic processing
return {
"primary_function": user_input,
"data_sources": [],
"technologies": [],
"category": "general",
"constraints": [],
"clarified_query": user_input
}
def suggest_refinements(self, query_analysis: Dict[str, Any]) -> List[str]:
"""Suggest query refinements to improve search results"""
system_message = """Based on the analyzed query, suggest 3-5 ways the user could refine their search to get better results.
Focus on:
- More specific technical requirements
- Additional context that might be helpful
- Alternative ways to phrase the requirement
- Related functionality they might also need
Return suggestions as a simple list, one per line."""
prompt = f"""
Query Analysis:
Primary Function: {query_analysis.get('primary_function', '')}
Data Sources: {', '.join(query_analysis.get('data_sources', []))}
Technologies: {', '.join(query_analysis.get('technologies', []))}
Category: {query_analysis.get('category', '')}
Suggest refinements:
"""
response = self.llm_client.generate_response(prompt, system_message)
suggestions = [s.strip() for s in response.split('\n') if s.strip()]
return suggestions[:5] # Limit to 5 suggestions
The QueryProcessor class provides sophisticated natural language understanding capabilities. The process_query method extracts structured information from conversational input, identifying key requirements that guide the search process.
The suggest_refinements method helps users improve their queries by providing specific suggestions for additional context or alternative phrasings that might yield better results.
Implementing Result Summarization and Presentation
The presentation layer transforms raw server information into clear, actionable summaries that help developers make informed decisions.
class ResultPresenter:
"""Formats and presents MCP server discovery results"""
def __init__(self, llm_client: LocalLLMClient):
self.llm_client = llm_client
def present_results(self, servers: List[MCPServerInfo], original_query: str) -> str:
"""Generate a comprehensive presentation of discovered servers"""
if not servers:
return "No MCP servers found matching your requirements. Try refining your search terms or exploring different categories."
# Generate summary introduction
intro = self._generate_introduction(servers, original_query)
# Format individual server details
server_details = []
for i, server in enumerate(servers, 1):
detail = self._format_server_detail(server, i)
server_details.append(detail)
# Generate recommendations
recommendations = self._generate_recommendations(servers, original_query)
# Combine all sections
result = f"""
{intro}
{'='*80}
DISCOVERED MCP SERVERS
{'='*80}
{chr(10).join(server_details)}
{'='*80}
RECOMMENDATIONS
{'='*80}
{recommendations}
{'='*80}
NEXT STEPS
{'='*80}
1. Review the server details above to identify the best match for your needs
2. Visit the provided URLs for complete documentation and installation instructions
3. Consider the provider reputation and maintenance status when making your selection
4. Test servers in a development environment before production deployment
5. Check compatibility with your specific MCP client implementation
For additional help, try refining your search with more specific requirements or explore related server categories.
"""
return result
def _generate_introduction(self, servers: List[MCPServerInfo], query: str) -> str:
"""Generate an introduction summarizing the search results"""
system_message = """Generate a brief introduction for MCP server search results.
Include:
- Number of servers found
- Main categories represented
- Brief assessment of how well they match the user's query
- Any notable patterns or highlights
Keep it concise and helpful."""
categories = list(set(server.category for server in servers))
providers = list(set(server.provider for server in servers))
prompt = f"""
Original Query: {query}
Search Results:
- {len(servers)} servers found
- Categories: {', '.join(categories)}
- Providers: {', '.join(providers[:5])}{'...' if len(providers) > 5 else ''}
Generate introduction:
"""
return self.llm_client.generate_response(prompt, system_message)
def _format_server_detail(self, server: MCPServerInfo, index: int) -> str:
"""Format detailed information for a single server"""
detail = f"""
{index}. {server.name}
{'-' * (len(str(index)) + 2 + len(server.name))}
DESCRIPTION:
{server.description}
FUNCTIONALITY:
{server.functionality}
DETAILS:
Category: {server.category}
Provider: {server.provider}
URL: {server.url}
INSTALLATION:
{server.installation_info}
"""
return detail
def _generate_recommendations(self, servers: List[MCPServerInfo], query: str) -> str:
"""Generate specific recommendations based on the discovered servers"""
system_message = """Based on the user's query and the discovered MCP servers, provide specific recommendations.
Include:
- Which server(s) best match the requirements
- Why those servers are recommended
- Any important considerations or trade-offs
- Suggestions for getting started
Be practical and actionable."""
server_summaries = [f"- {s.name}: {s.description}" for s in servers[:5]]
prompt = f"""
User Query: {query}
Top Servers Found:
{chr(10).join(server_summaries)}
Provide recommendations:
"""
return self.llm_client.generate_response(prompt, system_message)
The ResultPresenter class creates comprehensive, well-structured output that guides users through their options. The present_results method combines multiple information sources into a cohesive report that includes server details, analysis, and actionable recommendations.
Complete Working Example
Now we bring all components together into a fully functional MCP server discovery agent.
#!/usr/bin/env python3
"""
MCP Server Discovery Agent
A complete implementation of an LLM-powered agent for discovering
Model Context Protocol servers based on natural language queries.
Usage:
python mcp_discovery_agent.py
Requirements:
- ollama (with a model like mistral:7b installed)
- requests
- python 3.8+
Author: Tutorial Implementation
"""
import sys
import argparse
from typing import Optional
class MCPDiscoveryAgent:
"""Main agent class that orchestrates MCP server discovery"""
def __init__(self, model_name: str = "mistral:7b"):
"""Initialize the agent with specified LLM model"""
print(f"Initializing MCP Discovery Agent with model: {model_name}")
try:
self.llm_client = LocalLLMClient(model_name)
self.query_processor = QueryProcessor(self.llm_client)
self.discovery_engine = MCPServerDiscovery(self.llm_client)
self.result_presenter = ResultPresenter(self.llm_client)
print("Agent initialized successfully!")
except Exception as e:
print(f"Error initializing agent: {e}")
print("Please ensure Ollama is running and the specified model is available.")
sys.exit(1)
def search_servers(self, user_query: str, verbose: bool = False) -> str:
"""Main method to search for MCP servers based on user query"""
if verbose:
print(f"\nProcessing query: {user_query}")
try:
# Step 1: Process and analyze the query
if verbose:
print("Analyzing query requirements...")
query_analysis = self.query_processor.process_query(user_query)
if verbose:
print(f"Identified requirements: {query_analysis.get('primary_function', 'Unknown')}")
# Step 2: Discover relevant servers
if verbose:
print("Searching for relevant MCP servers...")
servers = self.discovery_engine.discover_servers(user_query)
if verbose:
print(f"Found {len(servers)} relevant servers")
# Step 3: Present results
if verbose:
print("Generating comprehensive report...")
result = self.result_presenter.present_results(servers, user_query)
return result
except Exception as e:
error_msg = f"Error during server discovery: {e}"
print(error_msg)
return error_msg
def interactive_mode(self):
"""Run the agent in interactive mode"""
print("\n" + "="*80)
print("MCP SERVER DISCOVERY AGENT")
print("="*80)
print("\nWelcome! I can help you find Model Context Protocol servers")
print("based on your requirements described in natural language.")
print("\nExamples:")
print("- 'I need a server for accessing PostgreSQL databases'")
print("- 'Find servers that can search web content'")
print("- 'I want to integrate with Google Drive'")
print("\nType 'quit' to exit, 'help' for more information.")
print("-"*80)
while True:
try:
user_input = input("\nWhat kind of MCP server are you looking for? ").strip()
if not user_input:
continue
if user_input.lower() in ['quit', 'exit', 'q']:
print("\nGoodbye! Happy building with MCP!")
break
if user_input.lower() in ['help', 'h']:
self._show_help()
continue
print("\nSearching for MCP servers...")
print("This may take a moment while I analyze your requirements and search multiple sources...")
result = self.search_servers(user_input, verbose=True)
print("\n" + result)
# Ask if user wants to refine the search
print("\n" + "-"*40)
refine = input("Would you like to refine your search? (y/n): ").strip().lower()
if refine in ['y', 'yes']:
suggestions = self.query_processor.suggest_refinements(
self.query_processor.process_query(user_input)
)
if suggestions:
print("\nHere are some suggestions to refine your search:")
for i, suggestion in enumerate(suggestions, 1):
print(f"{i}. {suggestion}")
print()
except KeyboardInterrupt:
print("\n\nExiting...")
break
except Exception as e:
print(f"\nError: {e}")
print("Please try again with a different query.")
def _show_help(self):
"""Display help information"""
help_text = """
HELP - MCP Server Discovery Agent
This agent helps you find Model Context Protocol (MCP) servers based on your requirements.
QUERY EXAMPLES:
- "I need a database server for MySQL"
- "Find servers for file system access"
- "Show me AI memory servers"
- "I want to integrate with Slack"
- "Search for web scraping servers"
TIPS FOR BETTER RESULTS:
- Be specific about your data sources (e.g., "PostgreSQL" vs "database")
- Mention specific platforms or tools you want to integrate with
- Include your use case context (e.g., "for building a chatbot")
- Specify any constraints (e.g., "open source only")
CATEGORIES OF MCP SERVERS:
- Data Storage (databases, file systems)
- Content Search (web search, document search)
- AI Memory (vector databases, knowledge graphs)
- Communication (Slack, email, messaging)
- Development Tools (Git, IDEs, testing)
- Cloud Infrastructure (AWS, Azure, GCP)
- Monitoring (logging, analytics, observability)
The agent searches both general web sources and specialized MCP directories
to find the most relevant servers for your needs.
"""
print(help_text)
def main():
"""Main entry point for the application"""
parser = argparse.ArgumentParser(
description="MCP Server Discovery Agent - Find Model Context Protocol servers using natural language"
)
parser.add_argument(
"--model",
default="mistral:7b",
help="Ollama model to use (default: mistral:7b)"
)
parser.add_argument(
"--query",
help="Single query to process (non-interactive mode)"
)
parser.add_argument(
"--verbose",
action="store_true",
help="Enable verbose output"
)
args = parser.parse_args()
# Initialize the agent
agent = MCPDiscoveryAgent(model_name=args.model)
if args.query:
# Single query mode
result = agent.search_servers(args.query, verbose=args.verbose)
print(result)
else:
# Interactive mode
agent.interactive_mode()
if __name__ == "__main__":
main()
This complete implementation provides both interactive and command-line interfaces for the MCP discovery agent. The interactive mode offers a conversational experience where users can refine their searches and explore suggestions, while the command-line mode enables integration into automated workflows.
Testing and Usage
To test the agent, ensure you have Ollama installed with an appropriate model. The agent works best with models that have good instruction-following capabilities such as Mistral 7B, Llama 2 7B, or larger models if your hardware supports them.
Start by running the agent in interactive mode to test basic functionality. Try queries like "I need a server for PostgreSQL database access" or "Find servers that can search web content" to verify that the search and analysis components work correctly.
The agent includes comprehensive error handling and fallback mechanisms to ensure robust operation even when individual components encounter issues. The modular design allows for easy extension and customization based on specific requirements.
For production use, consider implementing caching mechanisms to store search results and reduce redundant API calls. You might also want to add configuration files for customizing search sources and result formatting preferences.
The agent represents a powerful example of how local LLMs can be leveraged to create intelligent tools that understand natural language requirements and provide structured, actionable results. As the MCP ecosystem continues to grow, tools like this become increasingly valuable for helping developers navigate the expanding landscape of available integrations and capabilities.
No comments:
Post a Comment