Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_document.py:79-123 (handler)

  The get_document tool handler: async function that uses Elasticsearch client to retrieve a document by index and doc_id. Includes input schema via Annotated[Field], JSON output formatting, and detailed error handling for connection issues, missing index, missing document, etc.

  @app.tool(
      description="Retrieve a specific document from Elasticsearch index by document ID",
      tags={"elasticsearch", "get", "document", "retrieve"}
  )
  async def get_document(
          index: Annotated[str, Field(description="Name of the Elasticsearch index containing the document")],
          doc_id: Annotated[str, Field(description="Document ID to retrieve from the index")]
  ) -> str:
      """Retrieve a specific document from Elasticsearch index."""
      try:
          es = get_es_client()
  
          result = es.get(index=index, id=doc_id)
  
          return f"✅ Document retrieved successfully:\n\n{json.dumps(result, indent=2, ensure_ascii=False)}"
  
      except Exception as e:
          # Provide detailed error messages for different types of Elasticsearch errors
          error_message = "❌ Failed to get document:\n\n"
  
          error_str = str(e).lower()
          if "connection" in error_str or "refused" in error_str:
              error_message += "🔌 **Connection Error**: Cannot connect to Elasticsearch server\n"
              error_message += f"📍 Check if Elasticsearch is running at the configured address\n"
              error_message += f"💡 Try: Use 'setup_elasticsearch' tool to start Elasticsearch\n\n"
          elif (
                  "not_found" in error_str or "not found" in error_str) or "index_not_found_exception" in error_str or "no such index" in error_str:
              if "index" in error_str or "index_not_found_exception" in error_str or "no such index" in error_str:
                  error_message += f"📁 **Index Not Found**: Index '{index}' does not exist\n"
                  error_message += f"📍 The target index has not been created yet\n"
                  error_message += f"💡 **Suggestions for agents**:\n"
                  error_message += f"   1. Use 'list_indices' tool to see all available indices\n"
                  error_message += f"   2. Check which indices contain your target data\n"
                  error_message += f"   3. Use the correct index name from the list\n"
                  error_message += f"   4. If no suitable index exists, create one with 'create_index' tool\n\n"
              else:
                  error_message += f"📄 **Document Not Found**: Document ID '{doc_id}' does not exist\n"
                  error_message += f"📍 The requested document was not found in index '{index}'\n"
                  error_message += f"💡 Try: Check document ID or use 'search' to find documents\n\n"
          else:
              error_message += f"⚠️ **Unknown Error**: {str(e)}\n\n"
  
          error_message += f"🔍 **Technical Details**: {str(e)}"
  
          return error_message

• src/elasticsearch/elasticsearch_server.py:27-48 (registration)

  Mounts the elasticsearch_document sub-server into the unified Elasticsearch server app, registering the get_document tool (along with index_document and delete_document).

  from .sub_servers.elasticsearch_document import app as document_app
  from .sub_servers.elasticsearch_index import app as index_app
  from .sub_servers.elasticsearch_search import app as search_app
  from .sub_servers.elasticsearch_batch import app as batch_app
  
  # Create unified FastMCP application
  app = FastMCP(
      name="AgentKnowledgeMCP-Elasticsearch",
      version="2.0.0",
      instructions="Unified Elasticsearch tools for comprehensive knowledge management via modular server mounting"
  )
  
  # ================================
  # SERVER MOUNTING - MODULAR ARCHITECTURE
  # ================================
  
  print("🏗️ Mounting Elasticsearch sub-servers...")
  
  # Mount all sub-servers into unified interface
  app.mount(snapshots_app)           # 3 tools: snapshot management
  app.mount(index_metadata_app)      # 3 tools: metadata governance  
  app.mount(document_app)            # 3 tools: document operations

• src/main_server.py:19-76 (registration)

  Mounts the elasticsearch_server_app into the main FastMCP server, making get_document available at the top level.

  from src.elasticsearch.elasticsearch_server import app as elasticsearch_server_app  
  from src.prompts.prompt_server import app as prompt_server_app
  
  # Import middleware
  from src.middleware.confirmation_middleware import ConfirmationMiddleware
  
  # Load configuration and initialize components
  CONFIG = load_config()
  init_security(CONFIG["security"]["allowed_base_directory"])
  
  # Initialize confirmation manager
  confirmation_manager = initialize_confirmation_manager(CONFIG)
  print(f"✅ Confirmation system initialized (enabled: {CONFIG.get('confirmation', {}).get('enabled', True)})")
  
  # Auto-setup Elasticsearch if needed
  print("🔍 Checking Elasticsearch configuration...")
  config_path = Path(__file__).parent / "config.json"
  setup_result = auto_setup_elasticsearch(config_path, CONFIG)
  
  if setup_result["status"] == "setup_completed":
      # Reload config after setup
      CONFIG = load_config()
      print("✅ Elasticsearch auto-setup completed")
  elif setup_result["status"] == "already_configured":
      print("✅ Elasticsearch already configured")
  elif setup_result["status"] == "setup_failed":
      print(f"⚠️  Elasticsearch auto-setup failed: {setup_result.get('error', 'Unknown error')}")
      print("📝 You can manually setup using the 'setup_elasticsearch' tool")
  
  init_elasticsearch(CONFIG)
  
  # Create main FastMCP server
  app = FastMCP(
      name=CONFIG["server"]["name"],
      version=CONFIG["server"]["version"],
      instructions="🏗️ AgentKnowledgeMCP - Modern FastMCP server with modular composition architecture for knowledge management, Elasticsearch operations, file management, and system administration"
  )
  
  # ================================
  # MIDDLEWARE CONFIGURATION
  # ================================
  
  print("🔒 Adding confirmation middleware...")
  
  # Add confirmation middleware to main server
  app.add_middleware(ConfirmationMiddleware())
  
  print("✅ Confirmation middleware added successfully!")
  
  # ================================
  # SERVER COMPOSITION - MOUNTING
  # ================================
  
  print("🏗️ Mounting individual servers into main server...")
  
  # Mount Elasticsearch server with 'es' prefix
  # This provides: es_search, es_index_document, es_create_index, etc.
  app.mount(elasticsearch_server_app)

• src/elasticsearch/sub_servers/elasticsearch_document.py:84-85 (schema)

  Input schema for get_document tool: requires 'index' (str) and 'doc_id' (str) parameters with descriptions.

  index: Annotated[str, Field(description="Name of the Elasticsearch index containing the document")],
  doc_id: Annotated[str, Field(description="Document ID to retrieve from the index")]