Merge pull request #2 from PureCipher/feature/provenance-ledger

examples/ledger_example.py

@@ -0,0 +1,208 @@
+"""Example demonstrating the FastMCP Provenance Ledger functionality."""
+
+import asyncio
+from datetime import datetime
+from fastmcp import FastMCP
+from fastmcp.ledger import ProvenanceLedger, LedgerEvent, EventType
+
+
+def create_ledger_server():
+    """Create a FastMCP server with ledger functionality."""
+    server = FastMCP("LedgerExampleServer")
+    
+    # Enable the provenance ledger
+    ledger = server.enable_ledger(database_url="sqlite:///example_ledger.db")
+    
+    @server.tool
+    def log_tool_call(tool_name: str, parameters: dict, result: str) -> str:
+        """Log a tool call to the provenance ledger."""
+        # Create a ledger event
+        event = LedgerEvent(
+            event_type=EventType.TOOL_CALL,
+            actor_id="system",
+            resource_id=f"tool://{tool_name}",
+            action="execute",
+            metadata={
+                "tool_name": tool_name,
+                "parameters": parameters,
+                "result": result,
+                "timestamp": datetime.utcnow().isoformat()
+            }
+        )
+        
+        # Append to ledger
+        entry = ledger.append_event(event)
+        
+        return f"Logged tool call {tool_name} as entry {entry.sequence_number}"
+    
+    @server.tool
+    def log_policy_decision(policy_name: str, decision: str, context: dict) -> str:
+        """Log a policy decision to the provenance ledger."""
+        event = LedgerEvent(
+            event_type=EventType.POLICY_DECISION,
+            actor_id="policy_engine",
+            resource_id=f"policy://{policy_name}",
+            action="evaluate",
+            metadata={
+                "policy_name": policy_name,
+                "decision": decision,
+                "context": context,
+                "timestamp": datetime.utcnow().isoformat()
+            }
+        )
+        
+        entry = ledger.append_event(event)
+        return f"Logged policy decision {policy_name} as entry {entry.sequence_number}"
+    
+    @server.tool
+    def log_data_flow(source: str, destination: str, data_type: str, size: int) -> str:
+        """Log a data flow event to the provenance ledger."""
+        event = LedgerEvent(
+            event_type=EventType.DATA_FLOW,
+            actor_id="data_processor",
+            resource_id=f"data://{source}",
+            action="transfer",
+            metadata={
+                "source": source,
+                "destination": destination,
+                "data_type": data_type,
+                "size_bytes": size,
+                "timestamp": datetime.utcnow().isoformat()
+            }
+        )
+        
+        entry = ledger.append_event(event)
+        return f"Logged data flow from {source} to {destination} as entry {entry.sequence_number}"
+    
+    @server.tool
+    def verify_ledger_integrity() -> dict:
+        """Verify the integrity of the entire ledger."""
+        # Verify chain integrity
+        chain_valid = ledger.verify_chain_integrity()
+        
+        # Get statistics
+        stats = ledger.get_ledger_statistics()
+        
+        return {
+            "chain_integrity": chain_valid,
+            "statistics": stats,
+            "verification_timestamp": datetime.utcnow().isoformat()
+        }
+    
+    @server.tool
+    def get_ledger_entry(sequence_number: int) -> dict:
+        """Get a specific ledger entry."""
+        entry = ledger.get_entry(sequence_number)
+        
+        if not entry:
+            return {"error": f"Entry {sequence_number} not found"}
+        
+        event = entry.get_event()
+        
+        return {
+            "sequence_number": entry.sequence_number,
+            "entry_hash": entry.entry_hash,
+            "previous_hash": entry.previous_hash,
+            "created_at": entry.created_at.isoformat(),
+            "event": {
+                "type": event.event_type,
+                "actor_id": event.actor_id,
+                "action": event.action,
+                "metadata": event.metadata
+            }
+        }
+    
+    return server, ledger
+
+
+async def demonstrate_ledger_functionality():
+    """Demonstrate the ledger functionality."""
+    print("🚀 Creating FastMCP server with Provenance Ledger...")
+    server, ledger = create_ledger_server()
+    
+    print("\n📝 Logging various events to the ledger...")
+    
+    # Log some tool calls
+    print(server._tool_manager.call_tool("log_tool_call", {
+        "tool_name": "file_reader",
+        "parameters": {"path": "/data/file.txt"},
+        "result": "success"
+    }))
+    
+    print(server._tool_manager.call_tool("log_tool_call", {
+        "tool_name": "data_processor",
+        "parameters": {"input": "raw_data", "format": "json"},
+        "result": "processed_data"
+    }))
+    
+    # Log policy decisions
+    print(server._tool_manager.call_tool("log_policy_decision", {
+        "policy_name": "access_control",
+        "decision": "allow",
+        "context": {"user": "alice", "resource": "sensitive_data"}
+    }))
+    
+    print(server._tool_manager.call_tool("log_policy_decision", {
+        "policy_name": "data_retention",
+        "decision": "delete",
+        "context": {"age_days": 365, "type": "logs"}
+    }))
+    
+    # Log data flows
+    print(server._tool_manager.call_tool("log_data_flow", {
+        "source": "database",
+        "destination": "cache",
+        "data_type": "user_profiles",
+        "size": 1024
+    }))
+    
+    print(server._tool_manager.call_tool("log_data_flow", {
+        "source": "api",
+        "destination": "analytics",
+        "data_type": "usage_metrics",
+        "size": 512
+    }))
+    
+    print("\n🔍 Verifying ledger integrity...")
+    integrity_result = server._tool_manager.call_tool("verify_ledger_integrity", {})
+    print(f"Chain integrity: {integrity_result['chain_integrity']}")
+    print(f"Total entries: {integrity_result['statistics']['total_entries']}")
+    print(f"Total blocks: {integrity_result['statistics']['total_blocks']}")
+    
+    print("\n📋 Retrieving specific entries...")
+    for i in range(1, 4):
+        entry = server._tool_manager.call_tool("get_ledger_entry", {"sequence_number": i})
+        if "error" not in entry:
+            print(f"Entry {i}: {entry['event']['type']} - {entry['event']['action']}")
+    
+    print("\n🔗 Demonstrating hash chaining...")
+    entry1 = ledger.get_entry(1)
+    entry2 = ledger.get_entry(2)
+    
+    if entry1 and entry2:
+        print(f"Entry 1 hash: {entry1.entry_hash[:16]}...")
+        print(f"Entry 2 previous hash: {entry2.previous_hash[:16]}...")
+        print(f"Hash chain intact: {entry2.previous_hash == entry1.entry_hash}")
+    
+    print("\n🌳 Demonstrating Merkle tree verification...")
+    # Get the first block
+    block = ledger.get_block(1)
+    if block:
+        print(f"Block 1 Merkle root: {block.merkle_root[:16]}...")
+        print(f"Block 1 entry count: {block.entry_count}")
+        
+        # Verify block integrity
+        block_valid = ledger.verify_block_integrity(1)
+        print(f"Block 1 integrity: {block_valid}")
+    
+    print("\n✅ Ledger demonstration complete!")
+    print("\nThe ledger provides:")
+    print("- Tamper-evident hash chaining between entries")
+    print("- Merkle tree verification for blocks")
+    print("- Cryptographic integrity guarantees")
+    print("- Audit trail for all system events")
+    print("- HTTP API endpoints for external access")
+
+
+if __name__ == "__main__":
+    asyncio.run(demonstrate_ledger_functionality())

src/fastmcp/ledger/__init__.py

@@ -0,0 +1,17 @@
+"""FastMCP Ledger - Tamper-evident provenance ledger for auditability and non-repudiation."""
+
+from .ledger import LedgerEntry, LedgerBlock, ProvenanceLedger, LedgerEvent, EventType
+from .merkle import MerkleTree, MerkleProof
+from .adapter import LedgerAdapter, HyperledgerAdapter
+
+__all__ = [
+    "LedgerEntry", 
+    "LedgerBlock", 
+    "ProvenanceLedger", 
+    "LedgerEvent",
+    "EventType",
+    "MerkleTree", 
+    "MerkleProof",
+    "LedgerAdapter", 
+    "HyperledgerAdapter"
+]

src/fastmcp/ledger/adapter.py

@@ -0,0 +1,384 @@
+"""Ledger adapters for external blockchain backends like Hyperledger/OmniSeal."""
+
+import json
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List, Optional
+from datetime import datetime
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class LedgerAdapter(ABC):
+    """Abstract base class for ledger adapters."""
+    
+    @abstractmethod
+    async def submit_block(self, block_data: Dict[str, Any]) -> str:
+        """Submit a block to the external ledger.
+        
+        Args:
+            block_data: The block data to submit
+            
+        Returns:
+            Transaction ID or block hash from the external ledger
+        """
+        pass
+    
+    @abstractmethod
+    async def verify_block(self, block_id: str) -> bool:
+        """Verify a block exists and is valid on the external ledger.
+        
+        Args:
+            block_id: The block ID to verify
+            
+        Returns:
+            True if block is valid, False otherwise
+        """
+        pass
+    
+    @abstractmethod
+    async def get_block_proof(self, block_id: str) -> Optional[Dict[str, Any]]:
+        """Get a proof of block existence from the external ledger.
+        
+        Args:
+            block_id: The block ID to get proof for
+            
+        Returns:
+            Proof data or None if not found
+        """
+        pass
+
+
+class HyperledgerAdapter(LedgerAdapter):
+    """Adapter for Hyperledger Fabric blockchain backend."""
+    
+    def __init__(self, 
+                 network_config: str,
+                 channel_name: str = "mcp-channel",
+                 chaincode_name: str = "provenance-ledger",
+                 peer_endpoint: str = "localhost:7051",
+                 orderer_endpoint: str = "localhost:7050"):
+        """Initialize Hyperledger adapter.
+        
+        Args:
+            network_config: Path to network configuration file
+            channel_name: Name of the Hyperledger channel
+            chaincode_name: Name of the deployed chaincode
+            peer_endpoint: Peer endpoint URL
+            orderer_endpoint: Orderer endpoint URL
+        """
+        self.network_config = network_config
+        self.channel_name = channel_name
+        self.chaincode_name = chaincode_name
+        self.peer_endpoint = peer_endpoint
+        self.orderer_endpoint = orderer_endpoint
+        self._client = None
+        
+    async def _get_client(self):
+        """Get or create Hyperledger client."""
+        if self._client is None:
+            try:
+                # This would import the actual Hyperledger Fabric SDK
+                # from hfc.fabric import Client
+                # self._client = Client(net_profile=self.network_config)
+                logger.info("Hyperledger client initialized (stub implementation)")
+                self._client = "stub_client"
+            except ImportError:
+                logger.warning("Hyperledger Fabric SDK not available, using stub implementation")
+                self._client = "stub_client"
+        return self._client
+    
+    async def submit_block(self, block_data: Dict[str, Any]) -> str:
+        """Submit a block to Hyperledger Fabric.
+        
+        Args:
+            block_data: The block data to submit
+            
+        Returns:
+            Transaction ID from Hyperledger
+        """
+        try:
+            client = await self._get_client()
+            
+            # Prepare transaction data
+            transaction_data = {
+                "block_number": block_data.get("block_number"),
+                "merkle_root": block_data.get("merkle_root"),
+                "entry_count": block_data.get("entry_count"),
+                "timestamp": datetime.utcnow().isoformat(),
+                "entries": block_data.get("entries", [])
+            }
+            
+            # In a real implementation, this would:
+            # 1. Create a transaction proposal
+            # 2. Send it to endorsing peers
+            # 3. Submit to ordering service
+            # 4. Return transaction ID
+            
+            # Stub implementation
+            import hashlib
+            tx_data = json.dumps(transaction_data, sort_keys=True)
+            tx_id = hashlib.sha256(tx_data.encode()).hexdigest()
+            
+            logger.info(f"Submitted block {block_data.get('block_number')} to Hyperledger (tx: {tx_id})")
+            return tx_id
+            
+        except Exception as e:
+            logger.error(f"Failed to submit block to Hyperledger: {e}")
+            raise
+    
+    async def verify_block(self, block_id: str) -> bool:
+        """Verify a block exists on Hyperledger Fabric.
+        
+        Args:
+            block_id: The block ID to verify
+            
+        Returns:
+            True if block is valid, False otherwise
+        """
+        try:
+            client = await self._get_client()
+            
+            # In a real implementation, this would:
+            # 1. Query the blockchain for the block
+            # 2. Verify the block structure
+            # 3. Check block signatures
+            
+            # Stub implementation - always return True for demo
+            logger.info(f"Verified block {block_id} on Hyperledger")
+            return True
+            
+        except Exception as e:
+            logger.error(f"Failed to verify block {block_id}: {e}")
+            return False
+    
+    async def get_block_proof(self, block_id: str) -> Optional[Dict[str, Any]]:
+        """Get a proof of block existence from Hyperledger Fabric.
+        
+        Args:
+            block_id: The block ID to get proof for
+            
+        Returns:
+            Proof data or None if not found
+        """
+        try:
+            client = await self._get_client()
+            
+            # In a real implementation, this would:
+            # 1. Query the blockchain for block details
+            # 2. Get block header and signatures
+            # 3. Return proof data
+            
+            # Stub implementation
+            proof_data = {
+                "block_id": block_id,
+                "block_hash": f"hyperledger_hash_{block_id}",
+                "block_number": int(block_id.split('_')[-1]) if '_' in block_id else 0,
+                "timestamp": datetime.utcnow().isoformat(),
+                "proof_type": "hyperledger_fabric",
+                "signatures": ["peer1_signature", "peer2_signature"],
+                "merkle_root": f"merkle_root_{block_id}"
+            }
+            
+            logger.info(f"Retrieved block proof for {block_id} from Hyperledger")
+            return proof_data
+            
+        except Exception as e:
+            logger.error(f"Failed to get block proof for {block_id}: {e}")
+            return None
+
+
+class OmniSealAdapter(LedgerAdapter):
+    """Adapter for OmniSeal blockchain backend."""
+    
+    def __init__(self, 
+                 api_endpoint: str = "https://api.omniseal.com",
+                 api_key: str = None,
+                 network_id: str = "mainnet"):
+        """Initialize OmniSeal adapter.
+        
+        Args:
+            api_endpoint: OmniSeal API endpoint
+            api_key: API key for authentication
+            network_id: Network ID to use
+        """
+        self.api_endpoint = api_endpoint
+        self.api_key = api_key
+        self.network_id = network_id
+        self._session = None
+    
+    async def _get_session(self):
+        """Get or create HTTP session."""
+        if self._session is None:
+            try:
+                import aiohttp
+                headers = {}
+                if self.api_key:
+                    headers["Authorization"] = f"Bearer {self.api_key}"
+                self._session = aiohttp.ClientSession(
+                    base_url=self.api_endpoint,
+                    headers=headers
+                )
+            except ImportError:
+                logger.warning("aiohttp not available, using stub implementation")
+                self._session = "stub_session"
+        return self._session
+    
+    async def submit_block(self, block_data: Dict[str, Any]) -> str:
+        """Submit a block to OmniSeal.
+        
+        Args:
+            block_data: The block data to submit
+            
+        Returns:
+            Transaction ID from OmniSeal
+        """
+        try:
+            session = await self._get_session()
+            
+            # Prepare submission data
+            submission_data = {
+                "network_id": self.network_id,
+                "block_data": block_data,
+                "timestamp": datetime.utcnow().isoformat()
+            }
+            
+            # In a real implementation, this would:
+            # 1. Send POST request to OmniSeal API
+            # 2. Handle response and errors
+            # 3. Return transaction ID
+            
+            # Stub implementation
+            import hashlib
+            tx_data = json.dumps(submission_data, sort_keys=True)
+            tx_id = hashlib.sha256(tx_data.encode()).hexdigest()
+            
+            logger.info(f"Submitted block {block_data.get('block_number')} to OmniSeal (tx: {tx_id})")
+            return tx_id
+            
+        except Exception as e:
+            logger.error(f"Failed to submit block to OmniSeal: {e}")
+            raise
+    
+    async def verify_block(self, block_id: str) -> bool:
+        """Verify a block exists on OmniSeal.
+        
+        Args:
+            block_id: The block ID to verify
+            
+        Returns:
+            True if block is valid, False otherwise
+        """
+        try:
+            session = await self._get_session()
+            
+            # In a real implementation, this would:
+            # 1. Send GET request to OmniSeal API
+            # 2. Check response status
+            # 3. Verify block data
+            
+            # Stub implementation
+            logger.info(f"Verified block {block_id} on OmniSeal")
+            return True
+            
+        except Exception as e:
+            logger.error(f"Failed to verify block {block_id}: {e}")
+            return False
+    
+    async def get_block_proof(self, block_id: str) -> Optional[Dict[str, Any]]:
+        """Get a proof of block existence from OmniSeal.
+        
+        Args:
+            block_id: The block ID to get proof for
+            
+        Returns:
+            Proof data or None if not found
+        """
+        try:
+            session = await self._get_session()
+            
+            # In a real implementation, this would:
+            # 1. Send GET request to OmniSeal API
+            # 2. Parse response data
+            # 3. Return proof information
+            
+            # Stub implementation
+            proof_data = {
+                "block_id": block_id,
+                "block_hash": f"omniseal_hash_{block_id}",
+                "block_number": int(block_id.split('_')[-1]) if '_' in block_id else 0,
+                "timestamp": datetime.utcnow().isoformat(),
+                "proof_type": "omniseal",
+                "network_id": self.network_id,
+                "merkle_root": f"merkle_root_{block_id}"
+            }
+            
+            logger.info(f"Retrieved block proof for {block_id} from OmniSeal")
+            return proof_data
+            
+        except Exception as e:
+            logger.error(f"Failed to get block proof for {block_id}: {e}")
+            return None
+
+
+class StubAdapter(LedgerAdapter):
+    """Stub adapter for testing and development."""
+    
+    def __init__(self):
+        """Initialize stub adapter."""
+        self.submitted_blocks = {}
+        self.block_proofs = {}
+    
+    async def submit_block(self, block_data: Dict[str, Any]) -> str:
+        """Submit a block to the stub storage.
+        
+        Args:
+            block_data: The block data to submit
+            
+        Returns:
+            Generated block ID
+        """
+        import hashlib
+        block_id = f"stub_block_{len(self.submitted_blocks) + 1}"
+        self.submitted_blocks[block_id] = block_data
+        logger.info(f"Submitted block {block_data.get('block_number')} to stub storage (id: {block_id})")
+        return block_id
+    
+    async def verify_block(self, block_id: str) -> bool:
+        """Verify a block exists in stub storage.
+        
+        Args:
+            block_id: The block ID to verify
+            
+        Returns:
+            True if block exists, False otherwise
+        """
+        exists = block_id in self.submitted_blocks
+        logger.info(f"Verified block {block_id} in stub storage: {exists}")
+        return exists
+    
+    async def get_block_proof(self, block_id: str) -> Optional[Dict[str, Any]]:
+        """Get a proof of block existence from stub storage.
+        
+        Args:
+            block_id: The block ID to get proof for
+            
+        Returns:
+            Proof data or None if not found
+        """
+        if block_id not in self.submitted_blocks:
+            return None
+        
+        block_data = self.submitted_blocks[block_id]
+        proof_data = {
+            "block_id": block_id,
+            "block_hash": f"stub_hash_{block_id}",
+            "block_number": block_data.get("block_number", 0),
+            "timestamp": datetime.utcnow().isoformat(),
+            "proof_type": "stub",
+            "merkle_root": block_data.get("merkle_root", "")
+        }
+        
+        logger.info(f"Retrieved block proof for {block_id} from stub storage")
+        return proof_data

src/fastmcp/ledger/ledger.py

@@ -0,0 +1,504 @@
+"""Core ledger implementation with hash-linked entries and tamper-evident properties."""
+
+import hashlib
+import json
+from datetime import datetime
+from enum import Enum
+from typing import Any, Dict, List, Optional
+from uuid import UUID, uuid4
+
+from pydantic import BaseModel, Field
+from sqlmodel import SQLModel, Field as SQLField, Relationship
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class EventType(str, Enum):
+    """Types of events that can be recorded in the ledger."""
+    TOOL_CALL = "tool_call"
+    POLICY_DECISION = "policy_decision"
+    DATA_FLOW = "data_flow"
+    CONTRACT_ACTION = "contract_action"
+    AUTHENTICATION = "authentication"
+    AUTHORIZATION = "authorization"
+    SYSTEM_EVENT = "system_event"
+
+
+class LedgerEvent(BaseModel):
+    """A ledger event with structured data."""
+    
+    event_type: EventType = Field(..., description="Type of event")
+    actor_id: str = Field(..., description="ID of the actor performing the action")
+    resource_id: Optional[str] = Field(default=None, description="ID of the resource being acted upon")
+    action: str = Field(..., description="Action being performed")
+    metadata: Dict[str, Any] = Field(default_factory=dict, description="Additional event metadata")
+    timestamp: datetime = Field(default_factory=datetime.utcnow, description="Event timestamp")
+    data_hash: Optional[str] = Field(default=None, description="Hash of associated data")
+    
+    def get_content_hash(self) -> str:
+        """Get SHA-256 hash of event content for integrity verification."""
+        content = {
+            "event_type": self.event_type,
+            "actor_id": self.actor_id,
+            "resource_id": self.resource_id,
+            "action": self.action,
+            "metadata": self.metadata,
+            "data_hash": self.data_hash
+        }
+        content_str = json.dumps(content, sort_keys=True, default=str)
+        return hashlib.sha256(content_str.encode()).hexdigest()
+
+
+class LedgerEntry(SQLModel, table=True):
+    """A single entry in the provenance ledger with hash chaining."""
+    
+    __tablename__ = "ledger_entries"
+    
+    # Primary fields
+    id: UUID = SQLField(default_factory=uuid4, primary_key=True)
+    sequence_number: int = SQLField(..., index=True, description="Sequential entry number")
+    
+    # Event data
+    event_data: str = SQLField(..., description="JSON-encoded event data")
+    
+    # Hash chaining
+    previous_hash: Optional[str] = SQLField(default=None, description="Hash of previous entry")
+    entry_hash: str = SQLField(..., index=True, description="Hash of this entry")
+    
+    # Block information
+    block_id: Optional[UUID] = SQLField(default=None, foreign_key="ledger_blocks.id", index=True)
+    
+    # Timestamps
+    created_at: datetime = SQLField(default_factory=datetime.utcnow, index=True)
+    
+    # Verification
+    is_verified: bool = SQLField(default=True, description="Whether entry has been verified")
+    verification_timestamp: Optional[datetime] = SQLField(default=None)
+    
+    def get_event(self) -> LedgerEvent:
+        """Get parsed event from JSON."""
+        try:
+            event_data = json.loads(self.event_data)
+            return LedgerEvent(**event_data)
+        except (json.JSONDecodeError, ValueError) as e:
+            logger.error(f"Failed to parse event data: {e}")
+            raise ValueError(f"Invalid event data: {e}")
+    
+    def set_event(self, event: LedgerEvent) -> None:
+        """Set event as JSON."""
+        self.event_data = json.dumps(event.model_dump(), default=str)
+    
+    def calculate_hash(self) -> str:
+        """Calculate hash of this entry including previous hash."""
+        content = {
+            "sequence_number": self.sequence_number,
+            "event_data": self.event_data,
+            "previous_hash": self.previous_hash,
+            "created_at": self.created_at.isoformat()
+        }
+        content_str = json.dumps(content, sort_keys=True, default=str)
+        return hashlib.sha256(content_str.encode()).hexdigest()
+    
+    def verify_integrity(self) -> bool:
+        """Verify that the entry hash matches the calculated hash."""
+        calculated_hash = self.calculate_hash()
+        return calculated_hash == self.entry_hash
+
+
+class LedgerBlock(SQLModel, table=True):
+    """A block of ledger entries with Merkle tree root."""
+    
+    __tablename__ = "ledger_blocks"
+    
+    # Primary fields
+    id: UUID = SQLField(default_factory=uuid4, primary_key=True)
+    block_number: int = SQLField(..., index=True, description="Sequential block number")
+    
+    # Block metadata
+    entry_count: int = SQLField(..., description="Number of entries in this block")
+    first_entry_sequence: int = SQLField(..., description="Sequence number of first entry")
+    last_entry_sequence: int = SQLField(..., description="Sequence number of last entry")
+    
+    # Merkle tree
+    merkle_root: str = SQLField(..., description="Merkle tree root hash")
+    merkle_tree_data: str = SQLField(default="[]", description="JSON-encoded Merkle tree data")
+    
+    # Timestamps
+    created_at: datetime = SQLField(default_factory=datetime.utcnow, index=True)
+    sealed_at: Optional[datetime] = SQLField(default=None, description="When block was sealed")
+    
+    # Verification
+    is_verified: bool = SQLField(default=True, description="Whether block has been verified")
+    verification_timestamp: Optional[datetime] = SQLField(default=None)
+    
+    def get_merkle_tree_data(self) -> List[Dict[str, Any]]:
+        """Get parsed Merkle tree data from JSON."""
+        try:
+            return json.loads(self.merkle_tree_data)
+        except json.JSONDecodeError:
+            return []
+    
+    def set_merkle_tree_data(self, tree_data: List[Dict[str, Any]]) -> None:
+        """Set Merkle tree data as JSON."""
+        self.merkle_tree_data = json.dumps(tree_data)
+    
+    def verify_integrity(self, entries: List[LedgerEntry]) -> bool:
+        """Verify that the Merkle root matches the entries."""
+        if len(entries) != self.entry_count:
+            return False
+        
+        # Calculate Merkle root from entries
+        from .merkle import MerkleTree
+        entry_hashes = [entry.entry_hash for entry in entries]
+        merkle_tree = MerkleTree(entry_hashes)
+        calculated_root = merkle_tree.get_root()
+        
+        return calculated_root == self.merkle_root
+
+
+class ProvenanceLedger:
+    """Main ledger class for managing provenance entries with tamper-evident properties."""
+    
+    def __init__(self, database_url: str = "sqlite:///ledger.db"):
+        """Initialize the provenance ledger.
+        
+        Args:
+            database_url: Database connection URL
+        """
+        from sqlmodel import create_engine, Session
+        self.engine = create_engine(database_url, echo=False)
+        self._create_tables()
+        self._current_sequence = self._get_next_sequence_number()
+        self._current_block = None
+        self._block_size = 100  # Entries per block
+        
+    def _create_tables(self):
+        """Create database tables."""
+        try:
+            LedgerEntry.metadata.create_all(self.engine)
+            LedgerBlock.metadata.create_all(self.engine)
+            logger.info("Ledger database tables created/verified")
+        except Exception as e:
+            logger.error(f"Failed to create ledger tables: {e}")
+            raise
+    
+    def _get_next_sequence_number(self) -> int:
+        """Get the next sequence number for entries."""
+        from sqlmodel import Session, select, func
+        with Session(self.engine) as session:
+            result = session.exec(select(func.max(LedgerEntry.sequence_number))).first()
+            return (result or 0) + 1
+    
+    def _get_next_block_number(self) -> int:
+        """Get the next block number."""
+        from sqlmodel import Session, select, func
+        with Session(self.engine) as session:
+            result = session.exec(select(func.max(LedgerBlock.block_number))).first()
+            return (result or 0) + 1
+    
+    def append_event(self, event: LedgerEvent) -> LedgerEntry:
+        """Append a new event to the ledger.
+        
+        Args:
+            event: The event to append
+            
+        Returns:
+            The created ledger entry
+            
+        Raises:
+            ValueError: If event is invalid
+        """
+        try:
+            from sqlmodel import Session, select
+            
+            with Session(self.engine) as session:
+                # Get previous entry hash
+                previous_hash = None
+                if self._current_sequence > 1:
+                    prev_entry = session.exec(
+                        select(LedgerEntry).where(
+                            LedgerEntry.sequence_number == self._current_sequence - 1
+                        )
+                    ).first()
+                    if prev_entry:
+                        previous_hash = prev_entry.entry_hash
+                
+                # Create new entry
+                entry = LedgerEntry(
+                    sequence_number=self._current_sequence,
+                    previous_hash=previous_hash
+                )
+                entry.set_event(event)
+                entry.entry_hash = entry.calculate_hash()
+                
+                # Add to current block or create new block
+                if self._current_block is None or self._should_seal_block():
+                    self._current_block = self._create_new_block(session)
+                
+                entry.block_id = self._current_block.id
+                
+                # Save entry
+                session.add(entry)
+                session.commit()
+                session.refresh(entry)
+                
+                # Update block entry count
+                self._current_block.entry_count += 1
+                self._current_block.last_entry_sequence = entry.sequence_number
+                session.add(self._current_block)
+                
+                # Check if block should be sealed
+                if self._should_seal_block():
+                    self._seal_block(session)
+                
+                self._current_sequence += 1
+                
+                logger.info(f"Appended event {entry.id} to ledger (sequence: {entry.sequence_number})")
+                return entry
+                
+        except Exception as e:
+            logger.error(f"Failed to append event: {e}")
+            raise ValueError(f"Failed to append event: {e}")
+    
+    def _should_seal_block(self) -> bool:
+        """Check if current block should be sealed."""
+        return (self._current_block and 
+                self._current_block.entry_count >= self._block_size)
+    
+    def _create_new_block(self, session) -> LedgerBlock:
+        """Create a new block."""
+        block_number = self._get_next_block_number()
+        block = LedgerBlock(
+            block_number=block_number,
+            entry_count=0,
+            first_entry_sequence=self._current_sequence,
+            last_entry_sequence=self._current_sequence - 1,
+            merkle_root="",  # Will be set when sealed
+            merkle_tree_data="[]"
+        )
+        session.add(block)
+        session.commit()
+        session.refresh(block)
+        return block
+    
+    def _seal_block(self, session):
+        """Seal the current block with Merkle tree."""
+        if not self._current_block:
+            return
+        
+        from sqlmodel import select
+        
+        # Get all entries in this block
+        entries = session.exec(
+            select(LedgerEntry).where(
+                LedgerEntry.block_id == self._current_block.id
+            ).order_by(LedgerEntry.sequence_number)
+        ).all()
+        
+        if not entries:
+            return
+        
+        # Create Merkle tree
+        from .merkle import MerkleTree
+        entry_hashes = [entry.entry_hash for entry in entries]
+        merkle_tree = MerkleTree(entry_hashes)
+        
+        # Update block with Merkle root
+        self._current_block.merkle_root = merkle_tree.get_root()
+        self._current_block.merkle_tree_data = json.dumps(merkle_tree.get_tree_data())
+        self._current_block.sealed_at = datetime.utcnow()
+        self._current_block.is_verified = True
+        self._current_block.verification_timestamp = datetime.utcnow()
+        
+        session.add(self._current_block)
+        session.commit()
+        
+        logger.info(f"Sealed block {self._current_block.block_number} with {len(entries)} entries")
+        self._current_block = None
+
+    def seal_current_block(self) -> bool:
+        """Manually seal the current block if it exists.
+        
+        Returns:
+            True if block was sealed, False if no current block
+        """
+        if not self._current_block:
+            return False
+        
+        try:
+            from sqlmodel import Session
+            
+            with Session(self.engine) as session:
+                self._seal_block(session)
+            return True
+        except Exception as e:
+            logger.error(f"Failed to seal current block: {e}")
+            return False
+    
+    def get_entry(self, sequence_number: int) -> Optional[LedgerEntry]:
+        """Get a ledger entry by sequence number.
+        
+        Args:
+            sequence_number: The sequence number of the entry
+            
+        Returns:
+            The ledger entry or None if not found
+        """
+        from sqlmodel import Session, select
+        
+        with Session(self.engine) as session:
+            return session.exec(
+                select(LedgerEntry).where(
+                    LedgerEntry.sequence_number == sequence_number
+                )
+            ).first()
+    
+    def get_block(self, block_number: int) -> Optional[LedgerBlock]:
+        """Get a ledger block by block number.
+        
+        Args:
+            block_number: The block number
+            
+        Returns:
+            The ledger block or None if not found
+        """
+        from sqlmodel import Session, select
+        
+        with Session(self.engine) as session:
+            return session.exec(
+                select(LedgerBlock).where(
+                    LedgerBlock.block_number == block_number
+                )
+            ).first()
+    
+    def get_block_entries(self, block_number: int) -> List[LedgerEntry]:
+        """Get all entries in a block.
+        
+        Args:
+            block_number: The block number
+            
+        Returns:
+            List of ledger entries in the block
+        """
+        from sqlmodel import Session, select
+        
+        with Session(self.engine) as session:
+            block = self.get_block(block_number)
+            if not block:
+                return []
+            
+            return session.exec(
+                select(LedgerEntry).where(
+                    LedgerEntry.block_id == block.id
+                ).order_by(LedgerEntry.sequence_number)
+            ).all()
+    
+    def verify_block_integrity(self, block_number: int) -> bool:
+        """Verify the integrity of a block and its entries.
+        
+        Args:
+            block_number: The block number to verify
+            
+        Returns:
+            True if block is valid, False otherwise
+        """
+        try:
+            block = self.get_block(block_number)
+            if not block:
+                return False
+            
+            entries = self.get_block_entries(block_number)
+            if not entries:
+                return False
+            
+            # Verify each entry
+            for entry in entries:
+                if not entry.verify_integrity():
+                    logger.warning(f"Entry {entry.sequence_number} failed integrity check")
+                    return False
+            
+            # Verify block Merkle root
+            if not block.verify_integrity(entries):
+                logger.warning(f"Block {block_number} failed Merkle root verification")
+                return False
+            
+            return True
+            
+        except Exception as e:
+            logger.error(f"Failed to verify block {block_number}: {e}")
+            return False
+    
+    def verify_chain_integrity(self, start_sequence: int = 1, end_sequence: Optional[int] = None) -> bool:
+        """Verify the integrity of the entire chain or a range of entries.
+        
+        Args:
+            start_sequence: Starting sequence number (default: 1)
+            end_sequence: Ending sequence number (default: None for all)
+            
+        Returns:
+            True if chain is valid, False otherwise
+        """
+        try:
+            from sqlmodel import Session, select
+            
+            with Session(self.engine) as session:
+                query = select(LedgerEntry).where(
+                    LedgerEntry.sequence_number >= start_sequence
+                ).order_by(LedgerEntry.sequence_number)
+                
+                if end_sequence:
+                    query = query.where(LedgerEntry.sequence_number <= end_sequence)
+                
+                entries = session.exec(query).all()
+                
+                if not entries:
+                    return True
+                
+                # Verify each entry and hash chain
+                previous_hash = None
+                for entry in entries:
+                    # Verify entry integrity
+                    if not entry.verify_integrity():
+                        logger.warning(f"Entry {entry.sequence_number} failed integrity check")
+                        return False
+                    
+                    # Verify hash chain
+                    if previous_hash and entry.previous_hash != previous_hash:
+                        logger.warning(f"Hash chain broken at entry {entry.sequence_number}")
+                        return False
+                    
+                    previous_hash = entry.entry_hash
+                
+                return True
+                
+        except Exception as e:
+            logger.error(f"Failed to verify chain integrity: {e}")
+            return False
+    
+    def get_ledger_statistics(self) -> Dict[str, Any]:
+        """Get statistics about the ledger.
+        
+        Returns:
+            Dictionary with ledger statistics
+        """
+        from sqlmodel import Session, select, func
+        
+        with Session(self.engine) as session:
+            total_entries = session.exec(select(func.count(LedgerEntry.id))).first() or 0
+            total_blocks = session.exec(select(func.count(LedgerBlock.id))).first() or 0
+            sealed_blocks = session.exec(
+                select(func.count(LedgerBlock.id)).where(
+                    LedgerBlock.sealed_at.isnot(None)
+                )
+            ).first() or 0
+            
+            return {
+                "total_entries": total_entries,
+                "total_blocks": total_blocks,
+                "sealed_blocks": sealed_blocks,
+                "unsealed_blocks": total_blocks - sealed_blocks,
+                "current_sequence": self._current_sequence - 1,
+                "block_size": self._block_size
+            }

src/fastmcp/ledger/merkle.py

@@ -0,0 +1,244 @@
+"""Merkle tree implementation for ledger integrity verification."""
+
+import hashlib
+from typing import List, Dict, Any, Optional
+
+
+class MerkleProof:
+    """A Merkle proof for verifying an entry's inclusion in a Merkle tree."""
+    
+    def __init__(self, leaf_hash: str, path: List[Dict[str, str]], root_hash: str):
+        """Initialize a Merkle proof.
+        
+        Args:
+            leaf_hash: Hash of the leaf node being proven
+            path: List of path elements with 'hash' and 'position' ('left' or 'right')
+            root_hash: The root hash of the Merkle tree
+        """
+        self.leaf_hash = leaf_hash
+        self.path = path
+        self.root_hash = root_hash
+    
+    def verify(self) -> bool:
+        """Verify that this proof is valid.
+        
+        Returns:
+            True if the proof is valid, False otherwise
+        """
+        current_hash = self.leaf_hash
+        
+        for path_element in self.path:
+            sibling_hash = path_element['hash']
+            position = path_element['position']
+            
+            if position == 'left':
+                # Current hash is on the right, sibling on the left
+                combined = sibling_hash + current_hash
+            else:  # position == 'right'
+                # Current hash is on the left, sibling on the right
+                combined = current_hash + sibling_hash
+            
+            current_hash = hashlib.sha256(combined.encode()).hexdigest()
+        
+        return current_hash == self.root_hash
+
+
+class MerkleTree:
+    """A Merkle tree for efficient integrity verification of multiple entries."""
+    
+    def __init__(self, leaf_hashes: List[str]):
+        """Initialize a Merkle tree from a list of leaf hashes.
+        
+        Args:
+            leaf_hashes: List of SHA-256 hashes of the leaf nodes
+        """
+        if not leaf_hashes:
+            raise ValueError("Cannot create Merkle tree with empty leaf list")
+        
+        self.leaf_hashes = leaf_hashes.copy()
+        self.tree_data = []
+        self.root_hash = self._build_tree()
+    
+    def _build_tree(self) -> str:
+        """Build the Merkle tree and return the root hash.
+        
+        Returns:
+            The root hash of the Merkle tree
+        """
+        if len(self.leaf_hashes) == 1:
+            return self.leaf_hashes[0]
+        
+        # Ensure we have an even number of leaves by duplicating the last one if necessary
+        current_level = self.leaf_hashes.copy()
+        tree_levels = [current_level.copy()]
+        
+        while len(current_level) > 1:
+            next_level = []
+            
+            # Process pairs of nodes
+            for i in range(0, len(current_level), 2):
+                left_hash = current_level[i]
+                right_hash = current_level[i + 1] if i + 1 < len(current_level) else current_level[i]
+                
+                # Combine and hash
+                combined = left_hash + right_hash
+                parent_hash = hashlib.sha256(combined.encode()).hexdigest()
+                next_level.append(parent_hash)
+            
+            tree_levels.append(next_level.copy())
+            current_level = next_level
+        
+        # Store tree data for proof generation
+        self.tree_data = tree_levels
+        return current_level[0]
+    
+    def get_root(self) -> str:
+        """Get the root hash of the Merkle tree.
+        
+        Returns:
+            The root hash
+        """
+        return self.root_hash
+    
+    def get_tree_data(self) -> List[List[str]]:
+        """Get the complete tree data structure.
+        
+        Returns:
+            List of levels, where each level is a list of hashes
+        """
+        return self.tree_data
+    
+    def generate_proof(self, leaf_hash: str) -> Optional[MerkleProof]:
+        """Generate a Merkle proof for a specific leaf hash.
+        
+        Args:
+            leaf_hash: The hash of the leaf to prove
+            
+        Returns:
+            A MerkleProof object or None if the leaf is not found
+        """
+        try:
+            leaf_index = self.leaf_hashes.index(leaf_hash)
+        except ValueError:
+            return None
+        
+        if len(self.leaf_hashes) == 1:
+            # Single leaf case
+            return MerkleProof(leaf_hash, [], self.root_hash)
+        
+        path = []
+        current_index = leaf_index
+        
+        # Traverse up the tree
+        for level in range(len(self.tree_data) - 1):
+            current_level = self.tree_data[level]
+            
+            # Find sibling
+            if current_index % 2 == 0:  # Even index, sibling is on the right
+                sibling_index = current_index + 1
+                position = 'right'
+            else:  # Odd index, sibling is on the left
+                sibling_index = current_index - 1
+                position = 'left'
+            
+            # Add sibling to path if it exists
+            if sibling_index < len(current_level):
+                path.append({
+                    'hash': current_level[sibling_index],
+                    'position': position
+                })
+            
+            # Move to parent level
+            current_index = current_index // 2
+        
+        return MerkleProof(leaf_hash, path, self.root_hash)
+    
+    def verify_proof(self, proof: MerkleProof) -> bool:
+        """Verify a Merkle proof.
+        
+        Args:
+            proof: The MerkleProof to verify
+            
+        Returns:
+            True if the proof is valid, False otherwise
+        """
+        return proof.verify()
+    
+    def verify_leaf(self, leaf_hash: str) -> bool:
+        """Verify that a leaf hash is part of this Merkle tree.
+        
+        Args:
+            leaf_hash: The leaf hash to verify
+            
+        Returns:
+            True if the leaf is part of the tree, False otherwise
+        """
+        return leaf_hash in self.leaf_hashes
+    
+    def get_leaf_count(self) -> int:
+        """Get the number of leaf nodes in the tree.
+        
+        Returns:
+            The number of leaf nodes
+        """
+        return len(self.leaf_hashes)
+    
+    def get_leaf_hashes(self) -> List[str]:
+        """Get the list of leaf hashes.
+        
+        Returns:
+            List of leaf hashes
+        """
+        return self.leaf_hashes.copy()
+    
+    def get_tree_height(self) -> int:
+        """Get the height of the Merkle tree.
+        
+        Returns:
+            The height of the tree (number of levels)
+        """
+        return len(self.tree_data) if self.tree_data else 1
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert the Merkle tree to a dictionary representation.
+        
+        Returns:
+            Dictionary representation of the tree
+        """
+        return {
+            "root_hash": self.root_hash,
+            "leaf_count": self.get_leaf_count(),
+            "tree_height": self.get_tree_height(),
+            "tree_data": self.tree_data,
+            "leaf_hashes": self.leaf_hashes
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "MerkleTree":
+        """Create a MerkleTree from a dictionary representation.
+        
+        Args:
+            data: Dictionary representation of the tree
+            
+        Returns:
+            A MerkleTree instance
+        """
+        tree = cls(data["leaf_hashes"])
+        tree.tree_data = data["tree_data"]
+        tree.root_hash = data["root_hash"]
+        return tree
+
+
+def verify_merkle_proof(leaf_hash: str, proof_path: List[Dict[str, str]], root_hash: str) -> bool:
+    """Verify a Merkle proof without creating a MerkleTree instance.
+    
+    Args:
+        leaf_hash: Hash of the leaf being proven
+        proof_path: List of path elements with 'hash' and 'position'
+        root_hash: The expected root hash
+        
+    Returns:
+        True if the proof is valid, False otherwise
+    """
+    proof = MerkleProof(leaf_hash, proof_path, root_hash)
+    return proof.verify()

src/fastmcp/server/ledger_routes.py

@@ -0,0 +1,498 @@
+"""Ledger management HTTP routes."""
+
+from typing import Any, Dict, List, Optional
+from uuid import UUID
+
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.routing import Route
+
+from fastmcp.ledger import ProvenanceLedger, LedgerEvent, EventType, MerkleProof, LedgerBlock
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+async def append_event_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for appending events to the ledger.
+    
+    Expected JSON body:
+    {
+        "event_type": "tool_call",
+        "actor_id": "user123",
+        "resource_id": "resource456",
+        "action": "execute_tool",
+        "metadata": {...},
+        "data_hash": "sha256_hash_of_data"
+    }
+    """
+    try:
+        # Parse request body
+        body = await request.json()
+        
+        # Get ledger from request state
+        ledger: ProvenanceLedger = request.app.state.ledger
+        
+        # Create ledger event
+        event = LedgerEvent(**body)
+        
+        # Append to ledger
+        entry = ledger.append_event(event)
+        
+        return JSONResponse(
+            status_code=201,
+            content={
+                "entry_id": str(entry.id),
+                "sequence_number": entry.sequence_number,
+                "entry_hash": entry.entry_hash,
+                "block_id": str(entry.block_id) if entry.block_id else None,
+                "created_at": entry.created_at.isoformat()
+            }
+        )
+        
+    except ValueError as e:
+        logger.error(f"Invalid event data: {e}")
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid event data",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Failed to append event: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Failed to append event",
+                "reason": str(e)
+            }
+        )
+
+
+async def verify_block_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for verifying block integrity.
+    
+    Path parameter: block_number (int)
+    """
+    try:
+        # Get block number from path
+        block_number = int(request.path_params["block"])
+        
+        # Get ledger from request state
+        ledger: ProvenanceLedger = request.app.state.ledger
+        
+        # Verify block integrity
+        is_valid = ledger.verify_block_integrity(block_number)
+        
+        if not is_valid:
+            return JSONResponse(
+                status_code=400,
+                content={
+                    "block_number": block_number,
+                    "verified": False,
+                    "error": "Block integrity verification failed"
+                }
+            )
+        
+        # Get block details
+        block = ledger.get_block(block_number)
+        entries = ledger.get_block_entries(block_number)
+        
+        return JSONResponse(
+            status_code=200,
+            content={
+                "block_number": block_number,
+                "verified": True,
+                "block_id": str(block.id) if block else None,
+                "entry_count": len(entries),
+                "merkle_root": block.merkle_root if block else None,
+                "sealed_at": block.sealed_at.isoformat() if block and block.sealed_at else None,
+                "verification_timestamp": block.verification_timestamp.isoformat() if block and block.verification_timestamp else None
+            }
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid block number",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Failed to verify block: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Failed to verify block",
+                "reason": str(e)
+            }
+        )
+
+
+async def get_entry_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for getting a ledger entry by sequence number.
+    
+    Path parameter: sequence_number (int)
+    """
+    try:
+        # Get sequence number from path
+        sequence_number = int(request.path_params["sequence"])
+        
+        # Get ledger from request state
+        ledger: ProvenanceLedger = request.app.state.ledger
+        
+        # Get entry
+        entry = ledger.get_entry(sequence_number)
+        
+        if not entry:
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "error": "Entry not found",
+                    "sequence_number": sequence_number
+                }
+            )
+        
+        # Get event data
+        event = entry.get_event()
+        
+        return JSONResponse(
+            status_code=200,
+            content={
+                "entry_id": str(entry.id),
+                "sequence_number": entry.sequence_number,
+                "entry_hash": entry.entry_hash,
+                "previous_hash": entry.previous_hash,
+                "block_id": str(entry.block_id) if entry.block_id else None,
+                "created_at": entry.created_at.isoformat(),
+                "is_verified": entry.is_verified,
+                "event": {
+                    "event_type": event.event_type,
+                    "actor_id": event.actor_id,
+                    "resource_id": event.resource_id,
+                    "action": event.action,
+                    "metadata": event.metadata,
+                    "timestamp": event.timestamp.isoformat(),
+                    "data_hash": event.data_hash
+                }
+            }
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid sequence number",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Failed to get entry: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Failed to get entry",
+                "reason": str(e)
+            }
+        )
+
+
+async def get_block_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for getting a ledger block by block number.
+    
+    Path parameter: block_number (int)
+    """
+    try:
+        # Get block number from path
+        block_number = int(request.path_params["block"])
+        
+        # Get ledger from request state
+        ledger: ProvenanceLedger = request.app.state.ledger
+        
+        # Get block
+        block = ledger.get_block(block_number)
+        
+        if not block:
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "error": "Block not found",
+                    "block_number": block_number
+                }
+            )
+        
+        # Get entries in block
+        entries = ledger.get_block_entries(block_number)
+        
+        return JSONResponse(
+            status_code=200,
+            content={
+                "block_id": str(block.id),
+                "block_number": block.block_number,
+                "entry_count": block.entry_count,
+                "first_entry_sequence": block.first_entry_sequence,
+                "last_entry_sequence": block.last_entry_sequence,
+                "merkle_root": block.merkle_root,
+                "created_at": block.created_at.isoformat(),
+                "sealed_at": block.sealed_at.isoformat() if block.sealed_at else None,
+                "is_verified": block.is_verified,
+                "verification_timestamp": block.verification_timestamp.isoformat() if block.verification_timestamp else None,
+                "entries": [
+                    {
+                        "entry_id": str(entry.id),
+                        "sequence_number": entry.sequence_number,
+                        "entry_hash": entry.entry_hash,
+                        "created_at": entry.created_at.isoformat()
+                    }
+                    for entry in entries
+                ]
+            }
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid block number",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Failed to get block: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Failed to get block",
+                "reason": str(e)
+            }
+        )
+
+
+async def verify_chain_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for verifying chain integrity.
+    
+    Query parameters:
+    - start_sequence (int, optional): Starting sequence number (default: 1)
+    - end_sequence (int, optional): Ending sequence number (default: None)
+    """
+    try:
+        # Get query parameters
+        start_sequence = int(request.query_params.get("start_sequence", 1))
+        end_sequence = request.query_params.get("end_sequence")
+        if end_sequence:
+            end_sequence = int(end_sequence)
+        
+        # Get ledger from request state
+        ledger: ProvenanceLedger = request.app.state.ledger
+        
+        # Verify chain integrity
+        is_valid = ledger.verify_chain_integrity(start_sequence, end_sequence)
+        
+        return JSONResponse(
+            status_code=200,
+            content={
+                "verified": is_valid,
+                "start_sequence": start_sequence,
+                "end_sequence": end_sequence,
+                "verification_timestamp": ledger._get_next_sequence_number() - 1
+            }
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid sequence parameters",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Failed to verify chain: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Failed to verify chain",
+                "reason": str(e)
+            }
+        )
+
+
+async def get_merkle_proof_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for getting a Merkle proof for an entry.
+    
+    Path parameter: sequence_number (int)
+    """
+    try:
+        # Get sequence number from path
+        sequence_number = int(request.path_params["sequence"])
+        
+        # Get ledger from request state
+        ledger: ProvenanceLedger = request.app.state.ledger
+        
+        # Get entry
+        entry = ledger.get_entry(sequence_number)
+        
+        if not entry:
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "error": "Entry not found",
+                    "sequence_number": sequence_number
+                }
+            )
+        
+        # Get block - we need to find the block number from the entry
+        block = None
+        if entry.block_id:
+            # Find the block by querying for the block that contains this entry
+            from sqlmodel import Session, select
+            with Session(ledger.engine) as session:
+                block_query = session.exec(
+                    select(LedgerBlock).where(LedgerBlock.id == entry.block_id)
+                ).first()
+                if block_query:
+                    block = ledger.get_block(block_query.block_number)
+        
+        if not block:
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "error": "Block not found for entry",
+                    "sequence_number": sequence_number
+                }
+            )
+        
+        # Get all entries in block
+        entries = ledger.get_block_entries(block.block_number)
+        
+        # Create Merkle tree and generate proof
+        from fastmcp.ledger.merkle import MerkleTree
+        entry_hashes = [e.entry_hash for e in entries]
+        merkle_tree = MerkleTree(entry_hashes)
+        
+        proof = merkle_tree.generate_proof(entry.entry_hash)
+        
+        if not proof:
+            return JSONResponse(
+                status_code=500,
+                content={
+                    "error": "Failed to generate Merkle proof",
+                    "sequence_number": sequence_number
+                }
+            )
+        
+        return JSONResponse(
+            status_code=200,
+            content={
+                "sequence_number": sequence_number,
+                "entry_hash": entry.entry_hash,
+                "block_number": block.block_number,
+                "merkle_root": block.merkle_root,
+                "proof": {
+                    "leaf_hash": proof.leaf_hash,
+                    "path": proof.path,
+                    "root_hash": proof.root_hash
+                },
+                "verified": proof.verify()
+            }
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid sequence number",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Failed to get Merkle proof: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Failed to get Merkle proof",
+                "reason": str(e)
+            }
+        )
+
+
+async def get_ledger_statistics_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for getting ledger statistics."""
+    try:
+        # Get ledger from request state
+        ledger: ProvenanceLedger = request.app.state.ledger
+        
+        # Get statistics
+        stats = ledger.get_ledger_statistics()
+        
+        return JSONResponse(
+            status_code=200,
+            content=stats
+        )
+        
+    except Exception as e:
+        logger.error(f"Failed to get ledger statistics: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Failed to get ledger statistics",
+                "reason": str(e)
+            }
+        )
+
+
+def create_ledger_routes(ledger: ProvenanceLedger) -> List[Route]:
+    """Create ledger management routes.
+    
+    Args:
+        ledger: The provenance ledger instance
+        
+    Returns:
+        List of Starlette Route objects for ledger management
+    """
+    def endpoint_with_ledger(endpoint_func):
+        async def wrapper(request: Request) -> JSONResponse:
+            # Store ledger in app state for access in endpoint
+            request.app.state.ledger = ledger
+            return await endpoint_func(request)
+        return wrapper
+    
+    return [
+        Route(
+            path="/ledger/events",
+            endpoint=endpoint_with_ledger(append_event_endpoint),
+            methods=["POST"]
+        ),
+        Route(
+            path="/ledger/verify/{block}",
+            endpoint=endpoint_with_ledger(verify_block_endpoint),
+            methods=["GET"]
+        ),
+        Route(
+            path="/ledger/entries/{sequence}",
+            endpoint=endpoint_with_ledger(get_entry_endpoint),
+            methods=["GET"]
+        ),
+        Route(
+            path="/ledger/blocks/{block}",
+            endpoint=endpoint_with_ledger(get_block_endpoint),
+            methods=["GET"]
+        ),
+        Route(
+            path="/ledger/verify-chain",
+            endpoint=endpoint_with_ledger(verify_chain_endpoint),
+            methods=["GET"]
+        ),
+        Route(
+            path="/ledger/proof/{sequence}",
+            endpoint=endpoint_with_ledger(get_merkle_proof_endpoint),
+            methods=["GET"]
+        ),
+        Route(
+            path="/ledger/statistics",
+            endpoint=endpoint_with_ledger(get_ledger_statistics_endpoint),
+            methods=["GET"]
+        )
+    ]

src/fastmcp/server/server.py

@@ -64,6 +64,7 @@ from fastmcp.tools import ToolManager
 from fastmcp.tools.tool import FunctionTool, Tool, ToolResult
 from fastmcp.tools.tool_transform import ToolTransformConfig
 from fastmcp.policy import PolicyEngine
+from fastmcp.ledger import ProvenanceLedger
 from fastmcp.contracts import ContractEngine
 from fastmcp.utilities.cli import log_server_banner
 from fastmcp.utilities.components import FastMCPComponent
@@ -177,6 +178,7 @@ class FastMCP(Generic[LifespanResultT]):
         self._additional_http_routes: list[BaseRoute] = []
         self._mounted_servers: list[MountedServer] = []
         self._policy_engine: Optional[PolicyEngine] = None
+        self._ledger: Optional[ProvenanceLedger] = None
         self._contract_engine: Optional[ContractEngine] = None
         self._tool_manager = ToolManager(
             duplicate_behavior=on_duplicate_tools,
@@ -516,6 +518,13 @@ class FastMCP(Generic[LifespanResultT]):
             policy_route = create_policy_evaluate_route(self._policy_engine)
             routes.append(policy_route)
 
+        # Add ledger management endpoints if ledger is configured
+        if self._ledger is not None:
+            from fastmcp.server.ledger_routes import create_ledger_routes
+            
+            ledger_routes = create_ledger_routes(self._ledger)
+            routes.extend(ledger_routes)
+            
         # Add contract management endpoints if contract engine is configured
         if self._contract_engine is not None:
             from fastmcp.server.contract_routes import create_contract_routes
@@ -554,6 +563,31 @@ class FastMCP(Generic[LifespanResultT]):
         """
         return self._policy_engine
     
+    def enable_ledger(self, ledger: Optional[ProvenanceLedger] = None, database_url: str = "sqlite:///ledger.db") -> ProvenanceLedger:
+        """Enable the provenance ledger for this server.
+        
+        Args:
+            ledger: Optional ledger instance. If None, creates a new one.
+            database_url: Database URL for ledger persistence
+            
+        Returns:
+            The ledger instance
+        """
+        if ledger is None:
+            ledger = ProvenanceLedger(database_url)
+        
+        self._ledger = ledger
+        logger.info("Provenance ledger enabled for server")
+        return ledger
+    
+    def get_ledger(self) -> Optional[ProvenanceLedger]:
+        """Get the provenance ledger instance.
+        
+        Returns:
+            The ledger instance, or None if not enabled
+        """
+        return self._ledger
+
     def enable_contract_engine(self, contract_engine: Optional[ContractEngine] = None, database_url: str = "sqlite:///contracts.db") -> ContractEngine:
         """Enable the contract engine for this server.
         

tests/ledger/__init__.py

@@ -0,0 +1 @@
+"""Tests for the FastMCP Ledger module."""

tests/ledger/test_ledger_adapters.py

@@ -0,0 +1,234 @@
+"""Tests for the ledger adapters."""
+
+import pytest
+from fastmcp.ledger.adapter import HyperledgerAdapter, OmniSealAdapter, StubAdapter
+
+
+class TestHyperledgerAdapter:
+    """Test the HyperledgerAdapter class."""
+    
+    @pytest.fixture
+    def adapter(self):
+        """Create a Hyperledger adapter for testing."""
+        return HyperledgerAdapter(
+            network_config="test_config.json",
+            channel_name="test-channel",
+            chaincode_name="test-chaincode"
+        )
+    
+    async def test_adapter_initialization(self, adapter):
+        """Test adapter initialization."""
+        assert adapter.network_config == "test_config.json"
+        assert adapter.channel_name == "test-channel"
+        assert adapter.chaincode_name == "test-chaincode"
+        assert adapter.peer_endpoint == "localhost:7051"
+        assert adapter.orderer_endpoint == "localhost:7050"
+    
+    async def test_submit_block(self, adapter):
+        """Test block submission."""
+        block_data = {
+            "block_number": 1,
+            "merkle_root": "test_merkle_root",
+            "entry_count": 5,
+            "entries": ["entry1", "entry2", "entry3", "entry4", "entry5"]
+        }
+        
+        tx_id = await adapter.submit_block(block_data)
+        
+        assert tx_id is not None
+        assert len(tx_id) == 64  # SHA-256 hex length
+    
+    async def test_verify_block(self, adapter):
+        """Test block verification."""
+        block_id = "test_block_1"
+        
+        result = await adapter.verify_block(block_id)
+        
+        # Stub implementation always returns True
+        assert result is True
+    
+    async def test_get_block_proof(self, adapter):
+        """Test getting block proof."""
+        block_id = "test_block_1"
+        
+        proof = await adapter.get_block_proof(block_id)
+        
+        assert proof is not None
+        assert proof["block_id"] == block_id
+        assert "block_hash" in proof
+        assert "block_number" in proof
+        assert "timestamp" in proof
+        assert "proof_type" in proof
+        assert proof["proof_type"] == "hyperledger_fabric"
+        assert "signatures" in proof
+        assert "merkle_root" in proof
+
+
+class TestOmniSealAdapter:
+    """Test the OmniSealAdapter class."""
+    
+    @pytest.fixture
+    def adapter(self):
+        """Create an OmniSeal adapter for testing."""
+        return OmniSealAdapter(
+            api_endpoint="https://api.test.omniseal.com",
+            api_key="test_api_key",
+            network_id="testnet"
+        )
+    
+    async def test_adapter_initialization(self, adapter):
+        """Test adapter initialization."""
+        assert adapter.api_endpoint == "https://api.test.omniseal.com"
+        assert adapter.api_key == "test_api_key"
+        assert adapter.network_id == "testnet"
+    
+    async def test_submit_block(self, adapter):
+        """Test block submission."""
+        block_data = {
+            "block_number": 1,
+            "merkle_root": "test_merkle_root",
+            "entry_count": 3,
+            "entries": ["entry1", "entry2", "entry3"]
+        }
+        
+        tx_id = await adapter.submit_block(block_data)
+        
+        assert tx_id is not None
+        assert len(tx_id) == 64  # SHA-256 hex length
+    
+    async def test_verify_block(self, adapter):
+        """Test block verification."""
+        block_id = "test_block_1"
+        
+        result = await adapter.verify_block(block_id)
+        
+        # Stub implementation always returns True
+        assert result is True
+    
+    async def test_get_block_proof(self, adapter):
+        """Test getting block proof."""
+        block_id = "test_block_1"
+        
+        proof = await adapter.get_block_proof(block_id)
+        
+        assert proof is not None
+        assert proof["block_id"] == block_id
+        assert "block_hash" in proof
+        assert "block_number" in proof
+        assert "timestamp" in proof
+        assert "proof_type" in proof
+        assert proof["proof_type"] == "omniseal"
+        assert "network_id" in proof
+        assert proof["network_id"] == "testnet"
+        assert "merkle_root" in proof
+
+
+class TestStubAdapter:
+    """Test the StubAdapter class."""
+    
+    @pytest.fixture
+    def adapter(self):
+        """Create a stub adapter for testing."""
+        return StubAdapter()
+    
+    async def test_adapter_initialization(self, adapter):
+        """Test adapter initialization."""
+        assert adapter.submitted_blocks == {}
+        assert adapter.block_proofs == {}
+    
+    async def test_submit_block(self, adapter):
+        """Test block submission."""
+        block_data = {
+            "block_number": 1,
+            "merkle_root": "test_merkle_root",
+            "entry_count": 2,
+            "entries": ["entry1", "entry2"]
+        }
+        
+        block_id = await adapter.submit_block(block_data)
+        
+        assert block_id == "stub_block_1"
+        assert block_id in adapter.submitted_blocks
+        assert adapter.submitted_blocks[block_id] == block_data
+    
+    async def test_submit_multiple_blocks(self, adapter):
+        """Test submitting multiple blocks."""
+        for i in range(3):
+            block_data = {
+                "block_number": i + 1,
+                "merkle_root": f"merkle_root_{i}",
+                "entry_count": 1,
+                "entries": [f"entry_{i}"]
+            }
+            
+            block_id = await adapter.submit_block(block_data)
+            assert block_id == f"stub_block_{i + 1}"
+        
+        assert len(adapter.submitted_blocks) == 3
+    
+    async def test_verify_block(self, adapter):
+        """Test block verification."""
+        # Submit a block first
+        block_data = {"block_number": 1, "merkle_root": "test"}
+        block_id = await adapter.submit_block(block_data)
+        
+        # Verify existing block
+        result = await adapter.verify_block(block_id)
+        assert result is True
+        
+        # Verify non-existent block
+        result = await adapter.verify_block("nonexistent_block")
+        assert result is False
+    
+    async def test_get_block_proof(self, adapter):
+        """Test getting block proof."""
+        # Submit a block first
+        block_data = {
+            "block_number": 1,
+            "merkle_root": "test_merkle_root",
+            "entry_count": 2
+        }
+        block_id = await adapter.submit_block(block_data)
+        
+        # Get proof for existing block
+        proof = await adapter.get_block_proof(block_id)
+        
+        assert proof is not None
+        assert proof["block_id"] == block_id
+        assert proof["block_hash"] == f"stub_hash_{block_id}"
+        assert proof["block_number"] == 1
+        assert "timestamp" in proof
+        assert proof["proof_type"] == "stub"
+        assert proof["merkle_root"] == "test_merkle_root"
+        
+        # Get proof for non-existent block
+        proof = await adapter.get_block_proof("nonexistent_block")
+        assert proof is None
+    
+    async def test_adapter_state_persistence(self, adapter):
+        """Test that adapter maintains state across operations."""
+        # Submit a block
+        block_data = {"block_number": 1, "merkle_root": "test"}
+        block_id = await adapter.submit_block(block_data)
+        
+        # Verify it exists
+        assert await adapter.verify_block(block_id) is True
+        
+        # Get proof
+        proof = await adapter.get_block_proof(block_id)
+        assert proof is not None
+        
+        # Submit another block
+        block_data2 = {"block_number": 2, "merkle_root": "test2"}
+        block_id2 = await adapter.submit_block(block_data2)
+        
+        # Both blocks should exist
+        assert await adapter.verify_block(block_id) is True
+        assert await adapter.verify_block(block_id2) is True
+        
+        # Both proofs should be available
+        proof1 = await adapter.get_block_proof(block_id)
+        proof2 = await adapter.get_block_proof(block_id2)
+        assert proof1 is not None
+        assert proof2 is not None
+        assert proof1["block_id"] != proof2["block_id"]

tests/ledger/test_ledger_core.py

@@ -0,0 +1,305 @@
+"""Tests for the core ledger functionality."""
+
+import pytest
+from datetime import datetime
+from uuid import uuid4
+
+from fastmcp.ledger import ProvenanceLedger, LedgerEvent, EventType, LedgerEntry, LedgerBlock
+
+
+class TestProvenanceLedger:
+    """Test the ProvenanceLedger class."""
+    
+    @pytest.fixture
+    def ledger(self):
+        """Create a ledger instance for testing."""
+        return ProvenanceLedger("sqlite:///:memory:")
+    
+    @pytest.fixture
+    def sample_event(self):
+        """Create a sample event for testing."""
+        return LedgerEvent(
+            event_type=EventType.TOOL_CALL,
+            actor_id="user123",
+            resource_id="resource456",
+            action="execute_tool",
+            metadata={"tool_name": "test_tool", "parameters": {"x": 1, "y": 2}},
+            data_hash="sha256_hash_of_data"
+        )
+    
+    def test_ledger_initialization(self, ledger):
+        """Test ledger initialization."""
+        assert ledger is not None
+        assert ledger._current_sequence == 1
+        assert ledger._current_block is None
+        assert ledger._block_size == 100
+    
+    def test_append_event(self, ledger, sample_event):
+        """Test appending an event to the ledger."""
+        entry = ledger.append_event(sample_event)
+        
+        assert entry is not None
+        assert entry.sequence_number == 1
+        assert entry.previous_hash is None
+        assert entry.entry_hash is not None
+        assert entry.block_id is not None
+        assert entry.is_verified is True
+        
+        # Verify the event data
+        event = entry.get_event()
+        assert event.event_type == sample_event.event_type
+        assert event.actor_id == sample_event.actor_id
+        assert event.action == sample_event.action
+    
+    def test_hash_chaining(self, ledger, sample_event):
+        """Test that entries are properly hash-chained."""
+        # Append first event
+        entry1 = ledger.append_event(sample_event)
+        
+        # Append second event
+        entry2 = ledger.append_event(sample_event)
+        
+        # Verify hash chaining
+        assert entry2.previous_hash == entry1.entry_hash
+        assert entry2.sequence_number == entry1.sequence_number + 1
+    
+    def test_entry_integrity_verification(self, ledger, sample_event):
+        """Test entry integrity verification."""
+        entry = ledger.append_event(sample_event)
+        
+        # Verify entry integrity
+        assert entry.verify_integrity() is True
+        
+        # Tamper with the entry
+        original_hash = entry.entry_hash
+        entry.entry_hash = "tampered_hash"
+        
+        # Verify integrity fails
+        assert entry.verify_integrity() is False
+        
+        # Restore original hash
+        entry.entry_hash = original_hash
+        assert entry.verify_integrity() is True
+    
+    def test_get_entry(self, ledger, sample_event):
+        """Test retrieving an entry by sequence number."""
+        entry = ledger.append_event(sample_event)
+        
+        retrieved_entry = ledger.get_entry(entry.sequence_number)
+        assert retrieved_entry is not None
+        assert retrieved_entry.id == entry.id
+        assert retrieved_entry.sequence_number == entry.sequence_number
+    
+    def test_get_nonexistent_entry(self, ledger):
+        """Test retrieving a non-existent entry."""
+        entry = ledger.get_entry(999)
+        assert entry is None
+    
+    def test_block_creation_and_sealing(self, ledger, sample_event):
+        """Test block creation and sealing."""
+        # Append events to fill a block
+        entries = []
+        for i in range(5):  # Use a smaller number for testing
+            event = LedgerEvent(
+                event_type=EventType.TOOL_CALL,
+                actor_id=f"user{i}",
+                action=f"action_{i}",
+                metadata={"index": i}
+            )
+            entry = ledger.append_event(event)
+            entries.append(entry)
+        
+        # Manually seal the block (normally done when block_size is reached)
+        from sqlmodel import Session
+        with Session(ledger.engine) as session:
+            ledger._seal_block(session)
+        
+        # Get the block
+        block = ledger.get_block(1)
+        assert block is not None
+        assert block.block_number == 1
+        assert block.entry_count == 5
+        assert block.merkle_root is not None
+        assert block.sealed_at is not None
+        assert block.is_verified is True
+    
+    def test_block_integrity_verification(self, ledger, sample_event):
+        """Test block integrity verification."""
+        # Create a block with multiple entries
+        entries = []
+        for i in range(3):
+            event = LedgerEvent(
+                event_type=EventType.TOOL_CALL,
+                actor_id=f"user{i}",
+                action=f"action_{i}"
+            )
+            entry = ledger.append_event(event)
+            entries.append(entry)
+        
+        # Seal the block
+        from sqlmodel import Session
+        with Session(ledger.engine) as session:
+            ledger._seal_block(session)
+        
+        # Verify block integrity
+        assert ledger.verify_block_integrity(1) is True
+        
+        # Tamper with an entry
+        entry = ledger.get_entry(1)
+        original_hash = entry.entry_hash
+        entry.entry_hash = "tampered_hash"
+        
+        # Update the entry in the database
+        with Session(ledger.engine) as session:
+            session.add(entry)
+            session.commit()
+        
+        # Verify block integrity fails
+        assert ledger.verify_block_integrity(1) is False
+    
+    def test_chain_integrity_verification(self, ledger, sample_event):
+        """Test chain integrity verification."""
+        # Append multiple events
+        for i in range(5):
+            event = LedgerEvent(
+                event_type=EventType.TOOL_CALL,
+                actor_id=f"user{i}",
+                action=f"action_{i}"
+            )
+            ledger.append_event(event)
+        
+        # Verify entire chain
+        assert ledger.verify_chain_integrity() is True
+        
+        # Verify partial chain
+        assert ledger.verify_chain_integrity(start_sequence=2, end_sequence=4) is True
+    
+    def test_chain_integrity_with_tampering(self, ledger, sample_event):
+        """Test chain integrity verification with tampered entries."""
+        # Append multiple events
+        entries = []
+        for i in range(3):
+            event = LedgerEvent(
+                event_type=EventType.TOOL_CALL,
+                actor_id=f"user{i}",
+                action=f"action_{i}"
+            )
+            entry = ledger.append_event(event)
+            entries.append(entry)
+        
+        # Verify chain is intact
+        assert ledger.verify_chain_integrity() is True
+        
+        # Tamper with middle entry
+        middle_entry = entries[1]
+        original_hash = middle_entry.entry_hash
+        middle_entry.entry_hash = "tampered_hash"
+        
+        # Update in database
+        from sqlmodel import Session
+        with Session(ledger.engine) as session:
+            session.add(middle_entry)
+            session.commit()
+        
+        # Verify chain integrity fails
+        assert ledger.verify_chain_integrity() is False
+    
+    def test_ledger_statistics(self, ledger, sample_event):
+        """Test ledger statistics."""
+        # Initially empty
+        stats = ledger.get_ledger_statistics()
+        assert stats["total_entries"] == 0
+        assert stats["total_blocks"] == 0
+        assert stats["current_sequence"] == 0
+        
+        # Add some entries
+        for i in range(3):
+            event = LedgerEvent(
+                event_type=EventType.TOOL_CALL,
+                actor_id=f"user{i}",
+                action=f"action_{i}"
+            )
+            ledger.append_event(event)
+        
+        # Check updated statistics
+        stats = ledger.get_ledger_statistics()
+        assert stats["total_entries"] == 3
+        assert stats["total_blocks"] == 1
+        assert stats["current_sequence"] == 3
+    
+    def test_different_event_types(self, ledger):
+        """Test appending different types of events."""
+        event_types = [
+            EventType.TOOL_CALL,
+            EventType.POLICY_DECISION,
+            EventType.DATA_FLOW,
+            EventType.CONTRACT_ACTION,
+            EventType.AUTHENTICATION,
+            EventType.AUTHORIZATION,
+            EventType.SYSTEM_EVENT
+        ]
+        
+        for event_type in event_types:
+            event = LedgerEvent(
+                event_type=event_type,
+                actor_id="test_actor",
+                action="test_action"
+            )
+            entry = ledger.append_event(event)
+            
+            retrieved_event = entry.get_event()
+            assert retrieved_event.event_type == event_type
+    
+    def test_event_with_metadata(self, ledger):
+        """Test events with complex metadata."""
+        metadata = {
+            "nested": {"key": "value"},
+            "list": [1, 2, 3],
+            "boolean": True,
+            "null": None
+        }
+        
+        event = LedgerEvent(
+            event_type=EventType.TOOL_CALL,
+            actor_id="test_actor",
+            action="test_action",
+            metadata=metadata
+        )
+        
+        entry = ledger.append_event(event)
+        retrieved_event = entry.get_event()
+        
+        assert retrieved_event.metadata == metadata
+    
+    def test_event_content_hash(self, ledger):
+        """Test event content hash generation."""
+        event = LedgerEvent(
+            event_type=EventType.TOOL_CALL,
+            actor_id="test_actor",
+            action="test_action",
+            metadata={"key": "value"}
+        )
+        
+        content_hash = event.get_content_hash()
+        assert content_hash is not None
+        assert len(content_hash) == 64  # SHA-256 hex length
+        
+        # Same event should produce same hash
+        event2 = LedgerEvent(
+            event_type=EventType.TOOL_CALL,
+            actor_id="test_actor",
+            action="test_action",
+            metadata={"key": "value"}
+        )
+        
+        assert event2.get_content_hash() == content_hash
+        
+        # Different event should produce different hash
+        event3 = LedgerEvent(
+            event_type=EventType.TOOL_CALL,
+            actor_id="test_actor",
+            action="different_action",
+            metadata={"key": "value"}
+        )
+        
+        assert event3.get_content_hash() != content_hash

tests/ledger/test_ledger_http.py

@@ -0,0 +1,319 @@
+"""Tests for the ledger HTTP endpoints."""
+
+import pytest
+import httpx
+from fastmcp import FastMCP
+from fastmcp.ledger import ProvenanceLedger, LedgerEvent, EventType
+
+
+@pytest.fixture
+def server_with_ledger():
+    """Create a FastMCP server with ledger enabled."""
+    server = FastMCP("TestLedgerServer")
+    ledger = server.enable_ledger(database_url="sqlite:///:memory:")
+    return server, ledger
+
+
+@pytest.fixture
+async def client(server_with_ledger):
+    """Create an HTTP client for testing."""
+    server, ledger = server_with_ledger
+    from fastmcp.server.http import create_streamable_http_app
+    app = create_streamable_http_app(server, streamable_http_path="/")
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+        # Store server and ledger in client for test access
+        client.server = server
+        client.ledger = ledger
+        yield client
+
+
+class TestLedgerHTTPEndpoints:
+    """Test the ledger HTTP endpoints."""
+    
+    async def test_append_event_endpoint(self, client):
+        """Test the append event endpoint."""
+        event_data = {
+            "event_type": "tool_call",
+            "actor_id": "user123",
+            "resource_id": "resource456",
+            "action": "execute_tool",
+            "metadata": {"tool_name": "test_tool"},
+            "data_hash": "sha256_hash"
+        }
+        
+        response = await client.post("/ledger/events", json=event_data)
+        
+        assert response.status_code == 201
+        data = response.json()
+        assert "entry_id" in data
+        assert "sequence_number" in data
+        assert "entry_hash" in data
+        assert "block_id" in data
+        assert "created_at" in data
+        assert data["sequence_number"] == 1
+    
+    async def test_append_event_invalid_data(self, client):
+        """Test append event with invalid data."""
+        invalid_data = {
+            "event_type": "invalid_type",
+            "actor_id": "user123",
+            "action": "test_action"
+        }
+        
+        response = await client.post("/ledger/events", json=invalid_data)
+        
+        assert response.status_code == 400
+        data = response.json()
+        assert "error" in data
+        assert "Invalid event data" in data["error"]
+    
+    async def test_verify_block_endpoint(self, client):
+        """Test the verify block endpoint."""
+        # First, add some events to create a block
+        for i in range(3):
+            event_data = {
+                "event_type": "tool_call",
+                "actor_id": f"user{i}",
+                "action": f"action_{i}"
+            }
+            await client.post("/ledger/events", json=event_data)
+        
+        # Manually seal the block for testing
+        client.ledger.seal_current_block()
+    
+        # Verify the block
+        response = await client.get("/ledger/verify/1")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["block_number"] == 1
+        assert data["verified"] is True
+        assert "block_id" in data
+        assert "entry_count" in data
+        assert "merkle_root" in data
+    
+    async def test_verify_nonexistent_block(self, client):
+        """Test verifying a non-existent block."""
+        response = await client.get("/ledger/verify/999")
+        
+        assert response.status_code == 400
+        data = response.json()
+        assert "error" in data
+        assert "Block integrity verification failed" in data["error"]
+    
+    async def test_get_entry_endpoint(self, client):
+        """Test the get entry endpoint."""
+        # First, add an event
+        event_data = {
+            "event_type": "tool_call",
+            "actor_id": "user123",
+            "action": "test_action"
+        }
+        await client.post("/ledger/events", json=event_data)
+        
+        # Get the entry
+        response = await client.get("/ledger/entries/1")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["sequence_number"] == 1
+        assert "entry_hash" in data
+        assert "previous_hash" in data
+        assert "event" in data
+        assert data["event"]["actor_id"] == "user123"
+        assert data["event"]["action"] == "test_action"
+    
+    async def test_get_nonexistent_entry(self, client):
+        """Test getting a non-existent entry."""
+        response = await client.get("/ledger/entries/999")
+        
+        assert response.status_code == 404
+        data = response.json()
+        assert "error" in data
+        assert "Entry not found" in data["error"]
+    
+    async def test_get_block_endpoint(self, client):
+        """Test the get block endpoint."""
+        # First, add some events to create a block
+        for i in range(3):
+            event_data = {
+                "event_type": "tool_call",
+                "actor_id": f"user{i}",
+                "action": f"action_{i}"
+            }
+            await client.post("/ledger/events", json=event_data)
+        
+        # Get the block
+        response = await client.get("/ledger/blocks/1")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["block_number"] == 1
+        assert "entry_count" in data
+        assert "merkle_root" in data
+        assert "entries" in data
+        assert len(data["entries"]) == 3
+    
+    async def test_get_nonexistent_block(self, client):
+        """Test getting a non-existent block."""
+        response = await client.get("/ledger/blocks/999")
+        
+        assert response.status_code == 404
+        data = response.json()
+        assert "error" in data
+        assert "Block not found" in data["error"]
+    
+    async def test_verify_chain_endpoint(self, client):
+        """Test the verify chain endpoint."""
+        # Add some events
+        for i in range(3):
+            event_data = {
+                "event_type": "tool_call",
+                "actor_id": f"user{i}",
+                "action": f"action_{i}"
+            }
+            await client.post("/ledger/events", json=event_data)
+        
+        # Verify the chain
+        response = await client.get("/ledger/verify-chain")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["verified"] is True
+        assert "start_sequence" in data
+        assert "end_sequence" in data
+    
+    async def test_verify_chain_with_range(self, client):
+        """Test verify chain with specific range."""
+        # Add some events
+        for i in range(5):
+            event_data = {
+                "event_type": "tool_call",
+                "actor_id": f"user{i}",
+                "action": f"action_{i}"
+            }
+            await client.post("/ledger/events", json=event_data)
+        
+        # Verify partial chain
+        response = await client.get("/ledger/verify-chain?start_sequence=2&end_sequence=4")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["verified"] is True
+        assert data["start_sequence"] == 2
+        assert data["end_sequence"] == 4
+    
+    async def test_get_merkle_proof_endpoint(self, client):
+        """Test the get Merkle proof endpoint."""
+        # Add some events to create a block
+        for i in range(3):
+            event_data = {
+                "event_type": "tool_call",
+                "actor_id": f"user{i}",
+                "action": f"action_{i}"
+            }
+            await client.post("/ledger/events", json=event_data)
+        
+        # Manually seal the block for testing
+        client.ledger.seal_current_block()
+    
+        # Get Merkle proof for first entry
+        response = await client.get("/ledger/proof/1")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["sequence_number"] == 1
+        assert "entry_hash" in data
+        assert "block_number" in data
+        assert "merkle_root" in data
+        assert "proof" in data
+        assert "leaf_hash" in data["proof"]
+        assert "path" in data["proof"]
+        assert "root_hash" in data["proof"]
+        assert data["verified"] is True
+    
+    async def test_get_merkle_proof_nonexistent_entry(self, client):
+        """Test getting Merkle proof for non-existent entry."""
+        response = await client.get("/ledger/proof/999")
+        
+        assert response.status_code == 404
+        data = response.json()
+        assert "error" in data
+        assert "Entry not found" in data["error"]
+    
+    async def test_get_ledger_statistics_endpoint(self, client):
+        """Test the get ledger statistics endpoint."""
+        # Initially empty
+        response = await client.get("/ledger/statistics")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["total_entries"] == 0
+        assert data["total_blocks"] == 0
+        assert data["current_sequence"] == 0
+        
+        # Add some events
+        for i in range(3):
+            event_data = {
+                "event_type": "tool_call",
+                "actor_id": f"user{i}",
+                "action": f"action_{i}"
+            }
+            await client.post("/ledger/events", json=event_data)
+        
+        # Check updated statistics
+        response = await client.get("/ledger/statistics")
+        
+        assert response.status_code == 200
+        data = response.json()
+        assert data["total_entries"] == 3
+        assert data["total_blocks"] == 1
+        assert data["current_sequence"] == 3
+    
+    async def test_multiple_event_types(self, client):
+        """Test appending different event types."""
+        event_types = [
+            "tool_call",
+            "policy_decision", 
+            "data_flow",
+            "contract_action",
+            "authentication",
+            "authorization",
+            "system_event"
+        ]
+        
+        for event_type in event_types:
+            event_data = {
+                "event_type": event_type,
+                "actor_id": "test_actor",
+                "action": "test_action"
+            }
+            
+            response = await client.post("/ledger/events", json=event_data)
+            assert response.status_code == 201
+    
+    async def test_event_with_complex_metadata(self, client):
+        """Test event with complex metadata."""
+        event_data = {
+            "event_type": "tool_call",
+            "actor_id": "test_actor",
+            "action": "test_action",
+            "metadata": {
+                "nested": {"key": "value"},
+                "list": [1, 2, 3],
+                "boolean": True,
+                "null": None
+            }
+        }
+        
+        response = await client.post("/ledger/events", json=event_data)
+        
+        assert response.status_code == 201
+        data = response.json()
+        assert data["sequence_number"] == 1
+        
+        # Verify the event was stored correctly
+        entry_response = await client.get("/ledger/entries/1")
+        assert entry_response.status_code == 200
+        entry_data = entry_response.json()
+        assert entry_data["event"]["metadata"] == event_data["metadata"]

tests/ledger/test_merkle_tree.py

@@ -0,0 +1,235 @@
+"""Tests for the Merkle tree functionality."""
+
+import pytest
+
+from fastmcp.ledger.merkle import MerkleTree, MerkleProof, verify_merkle_proof
+
+
+class TestMerkleTree:
+    """Test the MerkleTree class."""
+    
+    def test_single_leaf(self):
+        """Test Merkle tree with a single leaf."""
+        leaf_hashes = ["hash1"]
+        tree = MerkleTree(leaf_hashes)
+        
+        assert tree.get_root() == "hash1"
+        assert tree.get_leaf_count() == 1
+        assert tree.get_tree_height() == 1
+    
+    def test_two_leaves(self):
+        """Test Merkle tree with two leaves."""
+        leaf_hashes = ["hash1", "hash2"]
+        tree = MerkleTree(leaf_hashes)
+        
+        # Root should be hash of concatenated leaves
+        import hashlib
+        expected_root = hashlib.sha256(("hash1" + "hash2").encode()).hexdigest()
+        assert tree.get_root() == expected_root
+        assert tree.get_leaf_count() == 2
+        assert tree.get_tree_height() == 2
+    
+    def test_three_leaves(self):
+        """Test Merkle tree with three leaves (odd number)."""
+        leaf_hashes = ["hash1", "hash2", "hash3"]
+        tree = MerkleTree(leaf_hashes)
+        
+        # With odd number of leaves, last leaf should be duplicated
+        assert tree.get_leaf_count() == 3
+        assert tree.get_tree_height() == 3
+    
+    def test_four_leaves(self):
+        """Test Merkle tree with four leaves."""
+        leaf_hashes = ["hash1", "hash2", "hash3", "hash4"]
+        tree = MerkleTree(leaf_hashes)
+        
+        assert tree.get_leaf_count() == 4
+        assert tree.get_tree_height() == 3
+    
+    def test_large_tree(self):
+        """Test Merkle tree with many leaves."""
+        leaf_hashes = [f"hash{i}" for i in range(100)]
+        tree = MerkleTree(leaf_hashes)
+        
+        assert tree.get_leaf_count() == 100
+        assert tree.get_tree_height() > 1
+        assert tree.get_root() is not None
+    
+    def test_generate_proof_single_leaf(self):
+        """Test proof generation for single leaf."""
+        leaf_hashes = ["hash1"]
+        tree = MerkleTree(leaf_hashes)
+        
+        proof = tree.generate_proof("hash1")
+        assert proof is not None
+        assert proof.leaf_hash == "hash1"
+        assert proof.path == []
+        assert proof.root_hash == "hash1"
+        assert proof.verify() is True
+    
+    def test_generate_proof_two_leaves(self):
+        """Test proof generation for two leaves."""
+        leaf_hashes = ["hash1", "hash2"]
+        tree = MerkleTree(leaf_hashes)
+        
+        # Proof for first leaf
+        proof1 = tree.generate_proof("hash1")
+        assert proof1 is not None
+        assert proof1.leaf_hash == "hash1"
+        assert len(proof1.path) == 1
+        assert proof1.path[0]["hash"] == "hash2"
+        assert proof1.path[0]["position"] == "right"
+        assert proof1.verify() is True
+        
+        # Proof for second leaf
+        proof2 = tree.generate_proof("hash2")
+        assert proof2 is not None
+        assert proof2.leaf_hash == "hash2"
+        assert len(proof2.path) == 1
+        assert proof2.path[0]["hash"] == "hash1"
+        assert proof2.path[0]["position"] == "left"
+        assert proof2.verify() is True
+    
+    def test_generate_proof_nonexistent_leaf(self):
+        """Test proof generation for non-existent leaf."""
+        leaf_hashes = ["hash1", "hash2"]
+        tree = MerkleTree(leaf_hashes)
+        
+        proof = tree.generate_proof("nonexistent")
+        assert proof is None
+    
+    def test_verify_leaf(self):
+        """Test leaf verification."""
+        leaf_hashes = ["hash1", "hash2", "hash3"]
+        tree = MerkleTree(leaf_hashes)
+        
+        # Verify existing leaves
+        assert tree.verify_leaf("hash1") is True
+        assert tree.verify_leaf("hash2") is True
+        assert tree.verify_leaf("hash3") is True
+        
+        # Verify non-existent leaf
+        assert tree.verify_leaf("nonexistent") is False
+    
+    def test_verify_proof(self):
+        """Test proof verification."""
+        leaf_hashes = ["hash1", "hash2", "hash3", "hash4"]
+        tree = MerkleTree(leaf_hashes)
+        
+        proof = tree.generate_proof("hash1")
+        assert proof is not None
+        assert tree.verify_proof(proof) is True
+        
+        # Tamper with proof
+        proof.leaf_hash = "tampered"
+        assert tree.verify_proof(proof) is False
+    
+    def test_to_dict_and_from_dict(self):
+        """Test serialization and deserialization."""
+        leaf_hashes = ["hash1", "hash2", "hash3"]
+        tree1 = MerkleTree(leaf_hashes)
+        
+        # Convert to dict
+        tree_dict = tree1.to_dict()
+        assert "root_hash" in tree_dict
+        assert "leaf_count" in tree_dict
+        assert "tree_height" in tree_dict
+        assert "tree_data" in tree_dict
+        assert "leaf_hashes" in tree_dict
+        
+        # Convert back to tree
+        tree2 = MerkleTree.from_dict(tree_dict)
+        assert tree2.get_root() == tree1.get_root()
+        assert tree2.get_leaf_count() == tree1.get_leaf_count()
+        assert tree2.get_tree_height() == tree1.get_tree_height()
+        assert tree2.get_leaf_hashes() == tree1.get_leaf_hashes()
+    
+    def test_empty_tree_raises_error(self):
+        """Test that empty tree raises error."""
+        with pytest.raises(ValueError, match="Cannot create Merkle tree with empty leaf list"):
+            MerkleTree([])
+
+
+class TestMerkleProof:
+    """Test the MerkleProof class."""
+    
+    def test_proof_verification_simple(self):
+        """Test simple proof verification."""
+        import hashlib
+        
+        # Create a simple two-leaf tree
+        leaf_hash = "hash1"
+        sibling_hash = "hash2"
+        combined = sibling_hash + leaf_hash  # sibling on left, leaf on right
+        root_hash = hashlib.sha256(combined.encode()).hexdigest()
+        
+        proof = MerkleProof(
+            leaf_hash=leaf_hash,
+            path=[{"hash": sibling_hash, "position": "left"}],
+            root_hash=root_hash
+        )
+        
+        assert proof.verify() is True
+    
+    def test_proof_verification_complex(self):
+        """Test complex proof verification with multiple levels."""
+        import hashlib
+        
+        # Create a more complex tree structure
+        leaf_hash = "hash1"
+        path = [
+            {"hash": "hash2", "position": "right"},  # sibling at leaf level
+            {"hash": "intermediate_hash", "position": "left"}  # sibling at parent level
+        ]
+        
+        # Calculate expected root
+        # First level: hash1 + hash2
+        level1 = hashlib.sha256(("hash1" + "hash2").encode()).hexdigest()
+        # Second level: intermediate_hash + level1
+        root_hash = hashlib.sha256(("intermediate_hash" + level1).encode()).hexdigest()
+        
+        proof = MerkleProof(
+            leaf_hash=leaf_hash,
+            path=path,
+            root_hash=root_hash
+        )
+        
+        assert proof.verify() is True
+    
+    def test_proof_verification_failure(self):
+        """Test proof verification failure."""
+        proof = MerkleProof(
+            leaf_hash="hash1",
+            path=[{"hash": "hash2", "position": "left"}],
+            root_hash="wrong_root_hash"
+        )
+        
+        assert proof.verify() is False
+
+
+class TestVerifyMerkleProof:
+    """Test the standalone verify_merkle_proof function."""
+    
+    def test_verify_merkle_proof_function(self):
+        """Test the standalone verify_merkle_proof function."""
+        import hashlib
+        
+        leaf_hash = "hash1"
+        path = [{"hash": "hash2", "position": "left"}]
+        # When sibling is on the left, we concatenate sibling + leaf
+        combined = "hash2" + leaf_hash
+        root_hash = hashlib.sha256(combined.encode()).hexdigest()
+        
+        assert verify_merkle_proof(leaf_hash, path, root_hash) is True
+        
+        # Test with wrong root
+        assert verify_merkle_proof(leaf_hash, path, "wrong_root") is False
+    
+    def test_verify_merkle_proof_empty_path(self):
+        """Test verify_merkle_proof with empty path (single leaf)."""
+        leaf_hash = "hash1"
+        path = []
+        root_hash = "hash1"  # For single leaf, root equals leaf
+        
+        assert verify_merkle_proof(leaf_hash, path, root_hash) is True
+        assert verify_merkle_proof(leaf_hash, path, "wrong_root") is False