Tribal Knowledge Service

Instructions

Implementation Reference

• src/mcp_server_tribal/mcp_app.py:93-146 (handler)

  Primary MCP tool handler for track_error: constructs ErrorRecord from parameters and stores it in ChromaDB via storage.add_error.
  @mcp.tool() async def track_error( error_type: str, error_message: str, language: str, framework: Optional[str] = None, code_snippet: Optional[str] = None, task_description: Optional[str] = None, solution_description: str = "", solution_code_fix: Optional[str] = None, solution_explanation: str = "", solution_references: Optional[List[str]] = None, ) -> Dict: """ Track an error and its solution in the knowledge base. Args: error_type: Type of error (e.g., ImportError, TypeError) error_message: The error message language: Programming language (e.g., python, javascript) framework: Framework used (e.g., fastapi, react) code_snippet: The code that caused the error task_description: What the user was trying to accomplish solution_description: Brief description of the solution solution_code_fix: Code that fixes the error solution_explanation: Detailed explanation of why the solution works solution_references: List of reference links Returns: The created error record """ if not solution_references: solution_references = [] error_data = ErrorRecord( error_type=error_type, context={ "language": language, "error_message": error_message, "framework": framework, "code_snippet": code_snippet, "task_description": task_description, }, solution={ "description": solution_description, "code_fix": solution_code_fix, "explanation": solution_explanation, "references": solution_references, }, ) error_record = await storage.add_error(error_data) return json.loads(error_record.model_dump_json())
• src/mcp_server_tribal/models/error_record.py:42-73 (schema)

  Pydantic model defining the input/output schema and validation for error records used by the track_error tool.
  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/chroma_storage.py:88-107 (helper)

  Helper method in ChromaStorage class that persists the ErrorRecord to ChromaDB collection.
  async def add_error(self, error: ErrorRecord) -> ErrorRecord: """Add a new error record to storage.""" document = self._error_to_document(error) # Store the document and metadata self.collection.add( ids=[str(error.id)], documents=[json.dumps(document)], metadatas=[ { "error_type": error.error_type, "language": error.context.language, "framework": error.context.framework or "", } ], # ChromaDB will auto-generate embeddings from the documents ) return error
• src/mcp_server_tribal/mcp_server.py:78-130 (handler)

  Secondary handler for track_error in proxy MCP server that sends data to the backend API endpoint.
  @mcp.tool() async def track_error( error_type: str, error_message: str, language: str, framework: Optional[str] = None, code_snippet: Optional[str] = None, task_description: Optional[str] = None, solution_description: str = "", solution_code_fix: Optional[str] = None, solution_explanation: str = "", solution_references: Optional[List[str]] = None, ) -> Dict: """ Track an error and its solution in the knowledge base. Args: error_type: Type of error (e.g., ImportError, TypeError) error_message: The error message language: Programming language (e.g., python, javascript) framework: Framework used (e.g., fastapi, react) code_snippet: The code that caused the error task_description: What the user was trying to accomplish solution_description: Brief description of the solution solution_code_fix: Code that fixes the error solution_explanation: Detailed explanation of why the solution works solution_references: List of reference links Returns: The created error record """ if not solution_references: solution_references = [] error_data = { "error_type": error_type, "context": { "language": language, "error_message": error_message, "framework": framework, "code_snippet": code_snippet, "task_description": task_description, }, "solution": { "description": solution_description, "code_fix": solution_code_fix, "explanation": solution_explanation, "references": solution_references, }, } return await make_api_request("POST", "/api/v1/errors/", data=error_data)
• src/mcp_server_tribal/mcp_server.py:213-239 (registration)

  Custom execution handler that registers and dispatches the track_error tool among others.
  @mcp.handle_execution async def handle_execution(tool_name: str, params: Dict) -> Dict: """ Handle tool execution. Args: tool_name: Name of the tool to execute params: Tool parameters Returns: Tool execution result """ logger.info(f"Executing tool: {tool_name} with params: {json.dumps(params)}") if tool_name == "track_error": return await track_error(**params) elif tool_name == "find_similar_errors": return await find_similar_errors(**params) elif tool_name == "search_errors": return await search_errors(**params) elif tool_name == "get_error_by_id": return await get_error_by_id(**params) elif tool_name == "get_api_status": return await get_api_status() else: logger.error(f"Unknown tool: {tool_name}") raise ValueError(f"Unknown tool: {tool_name}")