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