Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_document.py:375-471 (handler)

  Main FastMCP tool implementation for 'create_document_template'. Uses AI to enhance metadata, calls base helper, validates structure, and returns formatted JSON template with guidance.

  @app.tool(
      description="Create a properly structured document template for knowledge base with AI-generated metadata and formatting",
      tags={"elasticsearch", "document", "template", "knowledge-base", "ai-enhanced"}
  )
  async def create_document_template(
          title: Annotated[str, Field(description="Document title for the knowledge base entry")],
          content: Annotated[str, Field(description="Document content for AI analysis and metadata generation")] = "",
          priority: Annotated[
              str, Field(description="Priority level for the document", pattern="^(high|medium|low)$")] = "medium",
          source_type: Annotated[str, Field(description="Type of source content",
                                            pattern="^(markdown|code|config|documentation|tutorial)$")] = "markdown",
          tags: Annotated[
              List[str], Field(description="Additional manual tags (will be merged with AI-generated tags)")] = [],
          summary: Annotated[str, Field(description="Brief summary description of the document content")] = "",
          key_points: Annotated[List[str], Field(
              description="Additional manual key points (will be merged with AI-generated points)")] = [],
          related: Annotated[List[str], Field(description="List of related document IDs or references")] = [],
          use_ai_enhancement: Annotated[
              bool, Field(description="Use AI to generate intelligent tags and key points")] = True,
          ctx: Context = None
  ) -> str:
      """Create a properly structured document template for knowledge base indexing with AI-generated metadata."""
      try:
          # Initialize metadata
          final_tags = list(tags)  # Copy manual tags
          final_key_points = list(key_points)  # Copy manual key points
  
          # Use AI enhancement if requested and content is provided
          if use_ai_enhancement and content.strip() and ctx:
              try:
                  await ctx.info("🤖 Generating intelligent metadata and smart content using AI...")
                  ai_metadata = await generate_smart_metadata(title, content, ctx)
  
                  # Merge AI-generated tags with manual tags
                  ai_tags = ai_metadata.get("tags", [])
                  for tag in ai_tags:
                      if tag not in final_tags:
                          final_tags.append(tag)
  
                  # Merge AI-generated key points with manual points
                  ai_key_points = ai_metadata.get("key_points", [])
                  for point in ai_key_points:
                      if point not in final_key_points:
                          final_key_points.append(point)
  
                  # Use AI-generated smart summary if available
                  ai_summary = ai_metadata.get("smart_summary", "")
                  if ai_summary and not summary:
                      summary = ai_summary
  
                  # Use AI-enhanced content if available and better
                  ai_enhanced_content = ai_metadata.get("enhanced_content", "")
                  if ai_enhanced_content and len(ai_enhanced_content) > len(content) * 0.8:
                      content = ai_enhanced_content
  
                  await ctx.info(
                      f"✅ AI generated {len(ai_tags)} tags, {len(ai_key_points)} key points, smart summary, and enhanced content")
  
              except Exception as e:
                  await ctx.warning(f"AI enhancement failed: {str(e)}, using manual metadata only")
  
          # Generate auto-summary if not provided and content is available
          if not summary and content.strip():
              if len(content) > 200:
                  summary = content[:200].strip() + "..."
              else:
                  summary = content.strip()
  
          template = create_doc_template_base(
              title=title,
              priority=priority,
              source_type=source_type,
              tags=final_tags,
              summary=summary,
              key_points=final_key_points,
              related=related
          )
  
          ai_info = ""
          if use_ai_enhancement and ctx:
              ai_info = f"\n🤖 **AI Enhancement Used**: Generated {len(final_tags)} total tags and {len(final_key_points)} total key points\n"
  
          return (f"✅ Document template created successfully with AI-enhanced metadata!\n\n" +
                  f"{json.dumps(template, indent=2, ensure_ascii=False)}\n" +
                  ai_info +
                  f"\nThis template can be used with the 'index_document' tool.\n\n" +
                  f"⚠️ **CRITICAL: Search Before Creating - Avoid Duplicates**:\n" +
                  f"   🔍 **STEP 1**: Use 'search' tool to check if similar content already exists\n" +
                  f"   🔄 **STEP 2**: If found, UPDATE existing document instead of creating new one\n" +
                  f"   📝 **STEP 3**: For SHORT content (< 1000 chars): Add directly to 'content' field\n" +
                  f"   📁 **STEP 4**: For LONG content: Create file only when truly necessary\n" +
                  f"   🧹 **STEP 5**: Clean up outdated documents regularly to maintain quality\n" +
                  f"   🎯 **Remember**: Knowledge base quality > quantity - avoid bloat!")
  
      except Exception as e:
          return f"❌ Failed to create document template: {str(e)}"

• src/elasticsearch/document_schema.py:313-350 (helper)

  Base helper function imported as 'create_doc_template_base' and used by the main handler to generate the core document structure with ID generation and schema validation.

  def create_document_template(
      title: str,
      priority: str = "medium",
      source_type: str = "markdown",
      tags: Optional[List[str]] = None,
      summary: str = "",
      key_points: Optional[List[str]] = None,
      related: Optional[List[str]] = None
  ) -> Dict[str, Any]:
      """
      Create a document template with proper structure.
      
      Args:
          title: Document title
          priority: Priority level (high/medium/low)
          source_type: Type of source
          tags: List of tags
          summary: Brief description
          key_points: List of key points
          related: List of related document IDs
          
      Returns:
          Properly structured document
      """
      document = {
          "id": generate_document_id(title, source_type),
          "title": title,
          "summary": summary or f"Brief description of {title}",
          "content": "",  # Will be filled with actual content
          "last_modified": datetime.now().isoformat() + "Z",
          "priority": priority,
          "tags": tags or [],
          "related": related or [],
          "source_type": source_type,
          "key_points": key_points or []
      }
      
      return validate_document_structure(document)

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

  Mounting of the elasticsearch_document sub-server app into the main Elasticsearch FastMCP app, which registers the 'create_document_template' tool as part of the document operations.

  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
  app.mount(index_app)               # 3 tools: index management
  app.mount(search_app)              # 2 tools: search & validation
  app.mount(batch_app)               # 2 tools: batch operations

• src/elasticsearch/sub_servers/elasticsearch_document.py:380-394 (schema)

  Pydantic-based input schema definitions using Annotated[Type, Field(...)] for the tool parameters, providing validation, descriptions, patterns, and defaults.

  title: Annotated[str, Field(description="Document title for the knowledge base entry")],
  content: Annotated[str, Field(description="Document content for AI analysis and metadata generation")] = "",
  priority: Annotated[
      str, Field(description="Priority level for the document", pattern="^(high|medium|low)$")] = "medium",
  source_type: Annotated[str, Field(description="Type of source content",
                                    pattern="^(markdown|code|config|documentation|tutorial)$")] = "markdown",
  tags: Annotated[
      List[str], Field(description="Additional manual tags (will be merged with AI-generated tags)")] = [],
  summary: Annotated[str, Field(description="Brief summary description of the document content")] = "",
  key_points: Annotated[List[str], Field(
      description="Additional manual key points (will be merged with AI-generated points)")] = [],
  related: Annotated[List[str], Field(description="List of related document IDs or references")] = [],
  use_ai_enhancement: Annotated[
      bool, Field(description="Use AI to generate intelligent tags and key points")] = True,
  ctx: Context = None