Core Framework#

The Core Framework provides the essential foundation APIs that enable reliable, type-safe agentic system development. These five interconnected systems form the backbone of every capability, infrastructure node, and production deployment in the Osprey Framework.

System Architecture#

The Core Framework implements a Type-Safe, Convention-Driven Architecture with five integrated components:

🏗️ Base Components

Foundation & LangGraph Integration

BaseCapability, BaseInfrastructureNode, and decorators for seamless framework integration with error handling and planning.

Base Components

🔄 State & Context Management

Data Persistence & Exchange

AgentState with selective persistence, ContextManager for type-safe data access, and Pydantic-based serialization.

State and Context Management

📋 Registry System

Component Discovery & Management

Configuration-driven component loading with lazy initialization, dependency resolution, and type-safe access.

Registry System

⚙️ Configuration System

Environment & Settings Management

YAML-based configuration with environment resolution, model settings, and LangGraph integration.

Configuration System

💬 Prompt Management

Domain-Agnostic Prompt System

Dependency injection for prompt customization with builder patterns, defaults, and application-specific overrides.

Prompt System

⚡ Command System

Centralized Slash Commands

Unified command registry with categorization, autocompletion, and extensible handlers for all interfaces.

Command System

Framework Integration Patterns#

These systems work together to provide a unified development experience:

Startup and Initialization

How the framework initializes and loads components:

from osprey.registry import initialize_registry, get_registry
from osprey.state import StateManager

# 1. Initialize the global registry system (application startup)
initialize_registry()  # Loads all applications and components

# 2. Access the initialized registry
registry = get_registry()

# 3. Components are now available for use
capability = registry.get_capability('data_analysis')
context_class = registry.get_context_class('ANALYSIS_RESULTS')
data_source = registry.get_data_source('core_user_memory')

State and Context Flow

How state and context work together during execution:

from osprey.state import StateManager
from osprey.context import ContextManager

# 1. Create fresh state for new conversation
state = StateManager.create_fresh_state(
    user_input="Analyze beam performance data",
    current_state=previous_state  # Preserves context
)

# 2. Access context through ContextManager (manual pattern)
context = ContextManager(state)

# 3. Retrieve typed context objects
pv_data = context.get_context('PV_ADDRESSES', 'beam_current')

# 4. Within capabilities, use helper methods instead:
# contexts = self.get_required_contexts()
# return self.store_output_context(analysis_results)

Configuration Access

How to access configuration values throughout the framework:

from osprey.utils.config import (
    get_config_value, get_model_config, get_full_configuration
)

# Simple configuration access with defaults
timeout = get_config_value('execution.timeout', 30)
debug_mode = get_config_value('development.debug', False)

# Model-specific configuration
model_config = get_model_config('orchestrator')

# Full configuration for service passing
full_config = get_full_configuration()
service_config = {
    "configurable": {
        **full_config,
        "thread_id": f"my_service_{context_key}"
    }
}

Capability Implementation

Real capability implementation pattern from the framework:

from osprey.base import BaseCapability, capability_node
from osprey.utils.config import get_model_config
from applications.als_assistant.context_classes import AnalysisResultsContext

@capability_node
class DataAnalysisCapability(BaseCapability):
    name = "data_analysis"
    description = "General data analysis capability"
    provides = ["ANALYSIS_RESULTS"]
    requires = ["INPUT_DATA"]

    async def execute(self) -> Dict[str, Any]:
        # Get required contexts (automatically extracted)
        input_data, = self.get_required_contexts()

        # Get configuration for models/services
        model_config = get_model_config('python_code_generator')

        # Process data (actual implementation logic)
        analysis_results = await process_analysis(input_data)

        # Create typed context object
        result_context = AnalysisResultsContext(
            analysis_summary=analysis_results.summary,
            confidence_score=analysis_results.confidence,
            **analysis_results.data
        )

        # Store results in state (automatic context_key handling)
        return self.store_output_context(result_context)