Skip to content
Back to top

# Source code for mindroot.coreplugins.admin.persona_handler

    from pathlib import Path
    import json
    import logging
    from fastapi import HTTPException
    import shutil
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)

    def import_persona_from_index(index: str, persona: str):
        """Import a persona from the persona index.
        Args:
            index: Path to the persona index file
            persona: Name of the persona to import
        """
        index_path = Path('indices') / index / 'personas' / persona
        persona_path = Path('personas') / 'local' / persona
        shutil.copytree(index_path, persona_path)
        logger.info(f"Successfully imported persona '{persona}' from index '{index}'")

    def handle_persona_import(persona_data: dict, scope: str, owner: str=None) -> str:
        """Handle importing a persona from embedded data in agent configuration.
        Returns the persona name to be used in agent configuration.

        Args:
            persona_data: Dictionary containing persona data or string with persona name
            scope: 'local' or 'shared'

        Returns:
            str: Name of the persona to reference in agent config
        """
        if isinstance(persona_data, str):
            return persona_data
        if not isinstance(persona_data, dict):
            raise HTTPException(status_code=400, detail='Persona data must be either a string name or a dictionary')
        persona_name = persona_data.get('name')
        if not persona_name:
            raise HTTPException(status_code=400, detail='Persona name required in persona data')
        if owner and scope == 'registry':
            persona_path = Path(f'personas/registry/{owner}/{persona_name}/persona.json')
            return_name = f'registry/{owner}/{persona_name}'
        else:
            persona_path = Path('personas') / scope / persona_name / 'persona.json'
            return_name = persona_name
        if persona_path.exists():
            if owner and scope == 'registry':
                logger.info(f"Overwriting existing registry persona '{persona_name}' for owner '{owner}'")
            else:
                logger.warning(f"Persona '{persona_name}' already exists in {scope} scope - skipping import")
                return return_name
        try:
            persona_path.parent.mkdir(parents=True, exist_ok=True)
            with open(persona_path, 'w') as f:
                json.dump(persona_data, f, indent=2)
            logger.info(f"Successfully imported persona '{persona_name}' to {scope} scope")
            return return_name
        except Exception as e:
            logger.error(f"Failed to import persona '{persona_name}': {str(e)}")
            raise HTTPException(status_code=500, detail=f'Failed to import persona: {str(e)}')