Lobster AI - Cloud & Local Architecture

@dataclass
class AgentRegistryConfig:
    """Configuration for an agent in the system."""
    name: str                              # Unique agent identifier
    display_name: str                      # Human-readable name
    description: str                       # Agent's purpose/capability
    factory_function: str                  # Module path to factory function
    tier_requirement: str = 'free'         # 'free', 'premium', or 'enterprise'
    package_name: Optional[str] = None     # Package that provides this agent
    handoff_tool_name: Optional[str] = None     # Name of handoff tool
    handoff_tool_description: Optional[str] = None  # Tool description

# Example from lobster-research package
[project.entry-points."lobster.agents"]
data_expert_agent = "lobster_research.agents.data_expert:AGENT_CONFIG"
research_agent = "lobster_research.agents.research_agent:AGENT_CONFIG"

# Example from lobster-transcriptomics package
[project.entry-points."lobster.agents"]
transcriptomics_expert = "lobster_transcriptomics.agents.transcriptomics_expert:AGENT_CONFIG"

# lobster_research/agents/data_expert.py
AGENT_CONFIG = AgentRegistryConfig(
    name='data_expert_agent',
    display_name='Data Expert',
    description='Handles data fetching and download tasks',
    factory_function='lobster_research.agents.data_expert:data_expert',
    tier_requirement='free',
    package_name='lobster-research',
    handoff_tool_name='handoff_to_data_expert',
    handoff_tool_description='Assign data fetching/download tasks to the data expert'
)
        display_name='ML Expert',
        description='Handles Machine Learning related tasks like transforming the data in the desired format for downstream tasks',
        factory_function='lobster.agents.machine_learning_expert.machine_learning_expert',
        handoff_tool_name='handoff_to_machine_learning_expert',
        handoff_tool_description='Assign all machine learning related tasks (scVI, classification etc) to the machine learning expert agent'
    ),
    'visualization_expert_agent': AgentConfig(
        name='visualization_expert_agent',
        display_name='Visualization Expert',
        description='Creates publication-quality visualizations through supervisor-mediated workflows',
        factory_function='lobster.agents.visualization_expert.visualization_expert',
        handoff_tool_name='handoff_to_visualization_expert',
        handoff_tool_description='Delegate visualization tasks to the visualization expert agent'
    ),
    'proteomics_expert': AgentConfig(
        name='proteomics_expert',
        display_name='Proteomics Expert',
        description='Handles both mass spectrometry and affinity proteomics analysis tasks',
        factory_function='lobster.agents.proteomics_expert.proteomics_expert',
        handoff_tool_name='handoff_to_proteomics_expert',
        handoff_tool_description='Assign proteomics analysis tasks (mass spectrometry or affinity proteomics) to the proteomics expert'
    ),
\}

Adding new agents required updating:
├── lobster/agents/graph.py          # Import statements
├── lobster/agents/graph.py          # Agent creation code
├── lobster/agents/graph.py          # Handoff tool definitions
├── lobster/utils/callbacks.py       # Agent name hardcoded list
└── Multiple imports throughout codebase

Adding new agents only requires:
├── Define AGENT_CONFIG at module top
└── Register in pyproject.toml entry points

Everything else is handled automatically:
├── ✅ Dynamic agent discovery via ComponentRegistry
├── ✅ Automatic delegation tool creation
├── ✅ Callback system integration
├── ✅ Type-safe configuration
└── ✅ Professional error handling

# your_package/agents/new_agent.py
from lobster.core.registry import AgentRegistryConfig

# AGENT_CONFIG at module top for fast discovery
AGENT_CONFIG = AgentRegistryConfig(
    name='new_agent',
    display_name='New Agent',
    description='Handles specialized new functionality',
    factory_function='your_package.agents.new_agent:new_agent',
    tier_requirement='free',
    package_name='your-package',
    handoff_tool_name='handoff_to_new_agent',
    handoff_tool_description='Assign specialized tasks to the new agent'
)

def new_agent(data_manager, callback_handler=None, agent_name='new_agent', delegation_tools=None, workspace_path=None, **kwargs):
    """Create a new specialized agent."""
    # Agent implementation
    return agent_instance

# pyproject.toml
[project.entry-points."lobster.agents"]
new_agent = "your_package.agents.new_agent:AGENT_CONFIG"

# Get all worker agents with configurations
worker_agents = get_worker_agents()
# Returns: Dict[str, AgentConfig]

# Get all agent names (including system agents)
all_agents = get_all_agent_names()
# Returns: List[str]

# Get specific agent configuration
config = get_agent_config('data_expert_agent')
# Returns: AgentConfig or None

# Dynamically import agent factory
factory = import_agent_factory('lobster.agents.data_expert.data_expert')
# Returns: Callable

# tests/test_agent_discovery.py
from lobster.core.registry import ComponentRegistry
from lobster.testing import AgentContractTestMixin

def test_agent_discovery():
    """Test the agent discovery functionality."""
    registry = ComponentRegistry()

    # Test 1: Verify agents are discovered
    agents = registry.list_agents()
    assert len(agents) > 0

    # Test 2: Validate agent configs
    for agent_name in agents:
        config = registry.get_agent_config(agent_name)
        assert config is not None
        assert config.name == agent_name

    # Test 3: Check expected agents are present
    assert 'data_expert_agent' in agents
    assert 'transcriptomics_expert' in agents

lobster agents list  # Verify discovery
pytest tests/test_agent_discovery.py  # Run tests

# Primary concatenation method
concatenated_adata, statistics = concat_service.concatenate_samples(
    sample_adatas=sample_list,
    strategy=ConcatenationStrategy.SMART_SPARSE,
    batch_key="batch",
    use_intersecting_genes_only=True
)

# Concatenate from modality names
concatenated_adata, statistics = concat_service.concatenate_from_modalities(
    modality_names=["sample1", "sample2", "sample3"],
    output_name="concatenated_dataset",
    use_intersecting_genes_only=True
)

# Auto-detect samples by pattern
sample_modalities = concat_service.auto_detect_samples("geo_gse12345")

# Validate before processing
validation_result = concat_service.validate_concatenation_inputs(sample_list)

# Estimate memory requirements
memory_info = concat_service.estimate_memory_usage(sample_list)

@tool
def tool_name(modality_name: str, **params) -> str:
    """Professional tool with comprehensive error handling."""
    try:
        # 1. Validate modality exists
        if modality_name not in data_manager.list_modalities():
            raise ModalityNotFoundError(f"Modality '\{modality_name\}' not found")

        # 2. Get AnnData from modality
        adata = data_manager.get_modality(modality_name)

        # 3. Call stateless service
        result_adata, stats = service.method_name(adata, **params)

        # 4. Save new modality with descriptive name
        new_modality_name = f"\{modality_name\}_processed"
        data_manager.modalities[new_modality_name] = result_adata

        # 5. Log operation for provenance
        data_manager.log_tool_usage(tool_name, params, description)

        # 6. Format professional response
        return format_professional_response(stats, new_modality_name)

    except ServiceError as e:
        logger.error(f"Service error: \{e\}")
        return f"Service error: \{str(e)\}"
    except Exception as e:
        logger.error(f"Unexpected error: \{e\}")
        return f"Unexpected error: \{str(e)\}"

def service_method(
    self,
    adata: anndata.AnnData,
    **parameters
) -> Tuple[anndata.AnnData, Dict[str, Any]]:
    """
    Stateless service method working with AnnData directly.

    Returns:
        Tuple of (processed_adata, processing_statistics)

    """
    try:
        # 1. Create working copy
        adata_processed = adata.copy()

        # 2. Apply analysis algorithms
        # ... processing logic ...

        # 3. Calculate comprehensive statistics
        processing_stats = \{
            "analysis_type": "method_type",
            "parameters_used": parameters,
            "results": \{...\}
        \}

        return adata_processed, processing_stats

    except Exception as e:
        raise ServiceError(f"Method failed: \{str(e)\}")

geo_gse12345                    # Raw downloaded data
├── geo_gse12345_quality_assessed    # With QC metrics
├── geo_gse12345_filtered_normalized # Preprocessed data
├── geo_gse12345_doublets_detected   # With doublet annotations
├── geo_gse12345_clustered          # With clustering results
├── geo_gse12345_markers           # With marker genes
└── geo_gse12345_annotated        # With cell type annotations

1. check_data_status() → Review available modalities
2. assess_data_quality(modality_name) → Professional QC assessment
3. filter_and_normalize_modality(...) → Clean and normalize
4. detect_doublets_in_modality(...) → Remove doublets
5. cluster_modality(...) → Leiden clustering + UMAP
6. find_marker_genes_for_clusters(...) → Differential expression
7. annotate_cell_types(...) → Automated annotation
8. create_analysis_summary() → Comprehensive report

transcriptomics_expert.py: 450+ lines
├── Dual code paths (is_v2 checks everywhere)
├── Mixed responsibilities (orchestration + analysis)
├── Redundant implementations
├── Complex error handling
└── Maintenance overhead

transcriptomics_expert.py: 280 lines (clean)
├── Single DataManagerV2 path
├── Professional tool orchestration only
├── Stateless service delegation
├── Comprehensive error handling
└── Minimal maintenance overhead

Analysis Services: 4 refactored services
├── PreprocessingService: AnnData → (filtered_adata, stats)
├── QualityService: AnnData → (qc_adata, assessment)
├── ClusteringService: AnnData → (clustered_adata, results)
└── EnhancedSingleCellService: AnnData → (annotated_adata, metrics)

# Research Mode - Interactive exploration
SUPERVISOR_ASK_QUESTIONS=true
SUPERVISOR_WORKFLOW_GUIDANCE=detailed
# Result: 11K char prompt with full guidance

# Production Mode - Automated pipelines
SUPERVISOR_ASK_QUESTIONS=false
SUPERVISOR_WORKFLOW_GUIDANCE=minimal
# Result: 8K char prompt, 1.4K chars saved

# Development Mode - Debugging
SUPERVISOR_VERBOSE=true
SUPERVISOR_INCLUDE_SYSTEM=true
# Result: Detailed explanations with system info