Tribal Knowledge Service

Instructions

Implementation Reference

• src/mcp_server_tribal/mcp_app.py:164-200 (handler)

  Primary MCP tool handler for 'search_errors'. Constructs ErrorQuery from parameters and delegates to storage layer to retrieve matching error records.
  @mcp.tool() async def search_errors( error_type: Optional[str] = None, language: Optional[str] = None, framework: Optional[str] = None, error_message: Optional[str] = None, code_snippet: Optional[str] = None, task_description: Optional[str] = None, max_results: int = 5, ) -> List[Dict]: """ Search for errors in the knowledge base. Args: error_type: Type of error to filter by language: Programming language to filter by framework: Framework to filter by error_message: Error message to search for code_snippet: Code snippet to search for task_description: Task description to search for max_results: Maximum number of results to return Returns: List of matching error records """ query = ErrorQuery( error_type=error_type, language=language, framework=framework, error_message=error_message, code_snippet=code_snippet, task_description=task_description, max_results=max_results, ) records = await storage.search_errors(query) return [json.loads(record.model_dump_json()) for record in records]
• src/mcp_server_tribal/models/error_record.py:76-86 (schema)

  Pydantic model defining the input schema for search_errors query parameters.
  class ErrorQuery(BaseModel): """Query parameters for searching error records.""" error_type: Optional[str] = None language: Optional[str] = None framework: Optional[str] = None error_message: Optional[str] = None code_snippet: Optional[str] = None task_description: Optional[str] = None max_results: int = Field(default=5, ge=1, le=50)
• src/mcp_server_tribal/services/chroma_storage.py:165-211 (helper)

  ChromaDB storage backend implementation of search_errors, performing filtered semantic and metadata-based queries.
  async def search_errors(self, query: ErrorQuery) -> List[ErrorRecord]: """Search for error records based on the provided query.""" # Build metadata filter filter_clauses = [] if query.error_type: filter_clauses.append({"error_type": query.error_type}) if query.language: filter_clauses.append({"language": query.language}) if query.framework: filter_clauses.append({"framework": query.framework}) # Combine text for semantic search search_text = " ".join( [ query.error_message or "", query.code_snippet or "", query.task_description or "", ] ).strip() # If we have text to search, do a similarity search if search_text: results = self.collection.query( query_texts=[search_text], n_results=query.max_results, where=filter_clauses if filter_clauses else None, include=["documents"], ) else: # Otherwise, just get records matching the metadata filters results = self.collection.get( where=filter_clauses if filter_clauses else None, limit=query.max_results, include=["documents"], ) # Convert results to ErrorRecord objects error_records = [] if results.get("documents"): for doc_str in results["documents"][0]: document = json.loads(doc_str) error_records.append(self._document_to_error(document)) return error_records
• src/mcp_server_tribal/models/error_record.py:42-73 (schema)

  Pydantic model for ErrorRecord, defining the output schema structure for search results.
  class ErrorRecord(BaseModel): """Record of an error with context and solution.""" id: UUID = Field(default_factory=uuid4) error_type: str context: ErrorContext solution: ErrorSolution created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC)) metadata: dict = Field(default_factory=dict) schema_version: str = Field(default=SCHEMA_VERSION, description="Schema version for data migration") model_config = { "json_schema_extra": { "example": { "error_type": "ImportError", "context": { "language": "python", "framework": "fastapi", "error_message": "No module named 'fastapi'", "code_snippet": "from fastapi import FastAPI\napp = FastAPI()", "task_description": "Setting up a FastAPI server", }, "solution": { "description": "Install FastAPI package", "code_fix": "pip install fastapi", "explanation": "The fastapi package needs to be installed before importing it", "references": ["https://fastapi.tiangolo.com/tutorial/"], }, } } }
• src/mcp_server_tribal/services/storage_interface.py:78-88 (helper)

  Abstract method in storage interface defining the search_errors contract.
  async def search_errors(self, query: ErrorQuery) -> List[ErrorRecord]: """ Search for error records based on the provided query. Args: query: Search parameters Returns: A list of matching error records """ pass