Merge branch 'main' into feature/provenance-ledger

examples/contract_example.py

@@ -0,0 +1,401 @@
+"""
+FastMCP Contract Management Example
+
+This example demonstrates the inter-agent contract management system with:
+- Contract lifecycle management (draft, propose, sign, revoke)
+- Ed25519 cryptographic signatures for non-repudiation
+- HIPAA compliance support
+- SQLite persistence
+- HTTP API endpoints
+"""
+
+import asyncio
+import json
+from datetime import datetime, timedelta
+
+from fastmcp import FastMCP
+from fastmcp.contracts import ContractEngine, ContractState
+from fastmcp.contracts.contract import (
+    Clause, ContractCreateRequest, ContractProposeRequest,
+    ContractSignRequest, ContractRevokeRequest
+)
+from fastmcp.contracts.crypto import generate_key_pair, Ed25519Signer
+
+
+def create_contract_example():
+    """Create a FastMCP server with contract engine enabled."""
+
+    # Create server
+    server = FastMCP("Contract Management Server")
+
+    # Enable contract engine
+    contract_engine = server.enable_contract_engine()
+
+    # Add a simple tool that demonstrates contract operations
+    @server.tool
+    async def create_sample_contract(title: str, description: str) -> dict:
+        """Create a sample contract for demonstration."""
+
+        # Create sample clauses
+        clauses = [
+            Clause(
+                title="Data Protection",
+                content="All personal data must be protected according to HIPAA regulations.",
+                type="hipaa"
+            ),
+            Clause(
+                title="Access Control",
+                content="Only authorized personnel may access patient data.",
+                type="security"
+            ),
+            Clause(
+                title="Audit Trail",
+                content="All data access must be logged for audit purposes.",
+                type="compliance"
+            )
+        ]
+
+        # Create sample parties
+        parties = [
+            {
+                "id": "provider1",
+                "name": "Healthcare Provider Inc.",
+                "type": "provider",
+                "email": "provider@example.com",
+                "role": "data_controller"
+            },
+            {
+                "id": "patient1",
+                "name": "John Doe",
+                "type": "patient",
+                "email": "patient@example.com",
+                "role": "data_subject"
+            }
+        ]
+
+        # Create contract request
+        contract_request = ContractCreateRequest(
+            title=title,
+            description=description,
+            clauses=clauses,
+            parties=parties,
+            is_hipaa_compliant=True,
+            expires_at=datetime.utcnow() + timedelta(days=365),
+            metadata={
+                "created_by": "system",
+                "purpose": "data_sharing_agreement"
+            }
+        )
+
+        # Create contract
+        contract = await contract_engine.create_contract(contract_request, "system")
+
+        return {
+            "contract_id": str(contract.id),
+            "title": contract.title,
+            "state": contract.state.value,
+            "parties": [party["name"] for party in contract.get_parties()],
+            "clauses": [clause.title for clause in contract.get_clauses()],
+            "is_hipaa_compliant": contract.is_hipaa_compliant
+        }
+
+    return server
+
+
+async def demonstrate_contract_lifecycle():
+    """Demonstrate complete contract lifecycle."""
+
+    server = create_contract_example()
+    contract_engine = server.get_contract_engine()
+    assert contract_engine is not None  # Ensure contract engine is enabled
+
+    print("📋 FastMCP Contract Management Example")
+    print("=" * 50)
+
+    # 1. Create a contract
+    print("\n1️⃣ Creating Contract")
+    print("-" * 30)
+
+    clauses = [
+        Clause(
+            title="Data Sharing Agreement",
+            content="Healthcare provider may share patient data with authorized third parties.",
+            type="hipaa"
+        ),
+        Clause(
+            title="Consent Requirement",
+            content="Patient consent must be obtained before data sharing.",
+            type="consent"
+        )
+    ]
+
+    parties = [
+        {
+            "id": "provider1",
+            "name": "Metro Health System",
+            "type": "provider",
+            "email": "admin@metrohealth.com"
+        },
+        {
+            "id": "patient1",
+            "name": "Jane Smith",
+            "type": "patient",
+            "email": "jane.smith@email.com"
+        }
+    ]
+
+    contract_request = ContractCreateRequest(
+        title="HIPAA Data Sharing Agreement",
+        description="Agreement for sharing patient data between healthcare providers",
+        clauses=clauses,
+        parties=parties,
+        is_hipaa_compliant=True,
+        expires_at=datetime.utcnow() + timedelta(days=365)
+    )
+
+    contract = await contract_engine.create_contract(contract_request, "admin")
+    print(f"✅ Contract created: {contract.title}")
+    print(f"   ID: {contract.id}")
+    print(f"   State: {contract.state.value}")
+    print(f"   Parties: {len(contract.get_parties())}")
+    print(f"   HIPAA Compliant: {contract.is_hipaa_compliant}")
+
+    # 2. Propose the contract
+    print("\n2️⃣ Proposing Contract")
+    print("-" * 30)
+
+    proposal_request = ContractProposeRequest(
+        proposed_to=["provider1", "patient1"],
+        message="Please review and sign this data sharing agreement."
+    )
+
+    contract = await contract_engine.propose_contract(contract.id, proposal_request, "admin")
+    print(f"✅ Contract proposed to: {proposal_request.proposed_to}")
+    print(f"   State: {contract.state.value}")
+    print(f"   Proposed at: {contract.proposed_at}")
+
+    # 3. Sign the contract (Provider)
+    print("\n3️⃣ Signing Contract (Provider)")
+    print("-" * 30)
+
+    # Generate key pair for provider
+    provider_public_key, provider_private_key = generate_key_pair()
+    provider_signer = Ed25519Signer.from_private_key_b64(provider_private_key)
+
+    # Create signing message
+    signing_message = f"{contract.id}:{contract.get_content_hash()}:provider1:provider"
+    provider_signature = provider_signer.sign(signing_message)
+
+    sign_request = ContractSignRequest(
+        signer_id="provider1",
+        signer_type="provider",
+        public_key=provider_public_key,
+        signature=provider_signature
+    )
+
+    contract = await contract_engine.sign_contract(contract.id, sign_request)
+    print(f"✅ Provider signed the contract")
+    print(f"   Signatures: {len(contract.get_signatures())}")
+    print(f"   State: {contract.state.value}")
+
+    # 4. Sign the contract (Patient)
+    print("\n4️⃣ Signing Contract (Patient)")
+    print("-" * 30)
+
+    # Generate key pair for patient
+    patient_public_key, patient_private_key = generate_key_pair()
+    patient_signer = Ed25519Signer.from_private_key_b64(patient_private_key)
+
+    # Create signing message
+    signing_message = f"{contract.id}:{contract.get_content_hash()}:patient1:patient"
+    patient_signature = patient_signer.sign(signing_message)
+
+    sign_request = ContractSignRequest(
+        signer_id="patient1",
+        signer_type="patient",
+        public_key=patient_public_key,
+        signature=patient_signature
+    )
+
+    contract = await contract_engine.sign_contract(contract.id, sign_request)
+    print(f"✅ Patient signed the contract")
+    print(f"   Signatures: {len(contract.get_signatures())}")
+    print(f"   State: {contract.state.value}")
+    print(f"   Fully signed: {contract.is_fully_signed()}")
+
+    # 5. Demonstrate contract verification
+    print("\n5️⃣ Contract Verification")
+    print("-" * 30)
+
+    signatures = contract.get_signatures()
+    for signature in signatures:
+        # Verify signature
+        signing_message = f"{contract.id}:{contract.get_content_hash()}:{signature.signer_id}:{signature.signer_type}"
+        is_valid = provider_signer.verify(signing_message, signature.signature)
+        print(f"   {signature.signer_id} signature: {'✅ Valid' if is_valid else '❌ Invalid'}")
+
+    # 6. Get contract statistics
+    print("\n6️⃣ Contract Statistics")
+    print("-" * 30)
+
+    stats = await contract_engine.get_contract_statistics()
+    print(f"   Total contracts: {stats['total_contracts']}")
+    print(f"   By state: {stats['by_state']}")
+    print(f"   HIPAA compliant: {stats['hipaa_compliant']}")
+    print(f"   Signed contracts: {stats['signed_contracts']}")
+
+    # 7. Demonstrate revocation (optional)
+    print("\n7️⃣ Contract Revocation (Optional)")
+    print("-" * 30)
+
+    revoke_request = ContractRevokeRequest(
+        reason="Patient requested data deletion",
+        revoked_by="admin"
+    )
+
+    contract = await contract_engine.revoke_contract(contract.id, revoke_request)
+    print(f"✅ Contract revoked")
+    print(f"   State: {contract.state.value}")
+    print(f"   Revoked at: {contract.revoked_at}")
+    print(f"   Reason: {revoke_request.reason}")
+
+
+async def demonstrate_hipaa_compliance():
+    """Demonstrate HIPAA compliance features."""
+
+    print("\n🏥 HIPAA Compliance Demonstration")
+    print("=" * 50)
+
+    server = create_contract_example()
+    contract_engine = server.get_contract_engine()
+    assert contract_engine is not None
+
+    # Create HIPAA-compliant contract
+    hipaa_clauses = [
+        Clause(
+            title="HIPAA Privacy Rule",
+            content="All patient data must be handled according to HIPAA Privacy Rule requirements.",
+            type="hipaa",
+            metadata={"regulation": "45 CFR 164.502"}
+        ),
+        Clause(
+            title="Minimum Necessary Standard",
+            content="Only the minimum necessary information may be disclosed.",
+            type="hipaa",
+            metadata={"regulation": "45 CFR 164.502(b)"}
+        ),
+        Clause(
+            title="Business Associate Agreement",
+            content="Third parties must sign a Business Associate Agreement.",
+            type="hipaa",
+            metadata={"regulation": "45 CFR 164.502(e)"}
+        )
+    ]
+
+    hipaa_parties = [
+        {
+            "id": "covered_entity",
+            "name": "Regional Medical Center",
+            "type": "covered_entity",
+            "email": "privacy@regionalmedical.com",
+            "hipaa_role": "covered_entity"
+        },
+        {
+            "id": "business_associate",
+            "name": "HealthTech Solutions",
+            "type": "business_associate",
+            "email": "compliance@healthtech.com",
+            "hipaa_role": "business_associate"
+        }
+    ]
+
+    contract_request = ContractCreateRequest(
+        title="HIPAA Business Associate Agreement",
+        description="Agreement for HIPAA-compliant data processing services",
+        clauses=hipaa_clauses,
+        parties=hipaa_parties,
+        is_hipaa_compliant=True,
+        hipaa_entities=[
+            {
+                "type": "covered_entity",
+                "name": "Regional Medical Center",
+                "hipaa_id": "CE-001"
+            },
+            {
+                "type": "business_associate",
+                "name": "HealthTech Solutions",
+                "hipaa_id": "BA-001"
+            }
+        ],
+        metadata={
+            "hipaa_version": "2023",
+            "compliance_level": "full",
+            "audit_required": True
+        }
+    )
+
+    contract = await contract_engine.create_contract(contract_request, "hipaa_admin")
+    
+    print(f"✅ HIPAA-compliant contract created")
+    print(f"   Title: {contract.title}")
+    print(f"   HIPAA Compliant: {contract.is_hipaa_compliant}")
+    print(f"   HIPAA Entities: {len(contract.get_hipaa_entities())}")
+    print(f"   Clauses: {len(contract.get_clauses())}")
+
+    # Show HIPAA-specific metadata
+    metadata = contract.get_metadata()
+    print(f"   Compliance Level: {metadata.get('compliance_level')}")
+    print(f"   Audit Required: {metadata.get('audit_required')}")
+
+
+def main():
+    """Run the contract management example."""
+
+    async def run_example():
+        # Demonstrate contract lifecycle
+        await demonstrate_contract_lifecycle()
+
+        # Demonstrate HIPAA compliance
+        await demonstrate_hipaa_compliance()
+
+        print("\n🚀 Contract Management Example Complete!")
+        print("\nTo test the HTTP endpoints:")
+        print("1. Run: fastmcp run examples/contract_example.py --transport http")
+        print("2. Available endpoints:")
+        print("   - POST /contracts - Create contract")
+        print("   - GET /contracts - List contracts")
+        print("   - GET /contracts/{id} - Get contract")
+        print("   - POST /contracts/{id}/propose - Propose contract")
+        print("   - POST /contracts/{id}/sign - Sign contract")
+        print("   - POST /contracts/{id}/revoke - Revoke contract")
+        print("   - GET /contracts/statistics - Get statistics")
+        print("\n3. Example contract creation:")
+        print("""
+curl -X POST http://localhost:8000/contracts \\
+  -H "Content-Type: application/json" \\
+  -d '{
+    "title": "Test Contract",
+    "description": "A test contract",
+    "clauses": [
+      {
+        "title": "Test Clause",
+        "content": "This is a test clause",
+        "type": "test"
+      }
+    ],
+    "parties": [
+      {
+        "id": "party1",
+        "name": "Test Party",
+        "type": "provider"
+      }
+    ],
+    "is_hipaa_compliant": false
+  }'
+        """)
+
+    asyncio.run(run_example())
+
+
+if __name__ == "__main__":
+    main()

src/fastmcp/contracts/__init__.py

@@ -0,0 +1,7 @@
+"""FastMCP Contract Management - Inter-agent contract lifecycle with cryptographic guarantees."""
+
+from .contract import Contract, Clause, Signature, ContractState
+from .engine import ContractEngine
+from .registry import ContractRegistry
+
+__all__ = ["Contract", "Clause", "Signature", "ContractState", "ContractEngine", "ContractRegistry"]

src/fastmcp/contracts/contract.py

@@ -0,0 +1,306 @@
+"""Contract models and schemas for inter-agent contract management."""
+
+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, validator
+from sqlmodel import SQLModel, Field as SQLField, Relationship
+
+
+class ContractState(str, Enum):
+    """Contract lifecycle states."""
+    DRAFT = "draft"
+    PROPOSED = "proposed"
+    SIGNED = "signed"
+    REVOKED = "revoked"
+    EXPIRED = "expired"
+
+
+class Clause(BaseModel):
+    """A contract clause with structured content."""
+    
+    id: str = Field(default_factory=lambda: str(uuid4()))
+    title: str = Field(..., description="Clause title")
+    content: str = Field(..., description="Clause content")
+    type: str = Field(default="general", description="Clause type (e.g., 'hipaa', 'data_handling')")
+    metadata: Dict[str, Any] = Field(default_factory=dict, description="Additional clause metadata")
+    
+    class Config:
+        json_encoders = {
+            datetime: lambda v: v.isoformat()
+        }
+
+
+class Signature(BaseModel):
+    """Cryptographic signature for contract verification."""
+    
+    signer_id: str = Field(..., description="ID of the signing party")
+    signer_type: str = Field(..., description="Type of signer (e.g., 'provider', 'payor', 'patient')")
+    signature: str = Field(..., description="Base64-encoded Ed25519 signature")
+    public_key: str = Field(..., description="Base64-encoded public key")
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+    metadata: Dict[str, Any] = Field(default_factory=dict)
+    
+    class Config:
+        json_encoders = {
+            datetime: lambda v: v.isoformat()
+        }
+    
+    def model_dump(self, **kwargs):
+        """Override model_dump to handle datetime serialization."""
+        data = super().model_dump(**kwargs)
+        # Convert datetime objects to ISO format strings
+        for key, value in data.items():
+            if isinstance(value, datetime):
+                data[key] = value.isoformat()
+        return data
+
+
+class Contract(SQLModel, table=True):
+    """Contract model with SQLModel persistence."""
+    
+    __tablename__ = "contracts"
+    
+    # Primary fields
+    id: UUID = SQLField(default_factory=uuid4, primary_key=True)
+    title: str = SQLField(..., description="Contract title")
+    description: str = SQLField(..., description="Contract description")
+    
+    # Contract content
+    clauses: str = SQLField(..., description="JSON-encoded clauses")
+    parties: str = SQLField(..., description="JSON-encoded parties")
+    
+    # Lifecycle
+    state: ContractState = SQLField(default=ContractState.DRAFT)
+    created_at: datetime = SQLField(default_factory=datetime.utcnow)
+    proposed_at: Optional[datetime] = SQLField(default=None)
+    signed_at: Optional[datetime] = SQLField(default=None)
+    revoked_at: Optional[datetime] = SQLField(default=None)
+    expires_at: Optional[datetime] = SQLField(default=None)
+    
+    # Signatures
+    signatures: str = SQLField(default="[]", description="JSON-encoded signatures")
+    
+    # HIPAA compliance
+    is_hipaa_compliant: bool = SQLField(default=False)
+    hipaa_entities: str = SQLField(default="[]", description="JSON-encoded HIPAA entities")
+    
+    # Metadata
+    contract_metadata: str = SQLField(default="{}", description="JSON-encoded metadata")
+    version: str = SQLField(default="1.0.0")
+    
+    # Audit trail
+    created_by: str = SQLField(..., description="ID of the creating party")
+    last_modified: datetime = SQLField(default_factory=datetime.utcnow)
+    
+    def get_clauses(self) -> List[Clause]:
+        """Get parsed clauses from JSON."""
+        try:
+            clauses_data = json.loads(self.clauses)
+            return [Clause(**clause) for clause in clauses_data]
+        except (json.JSONDecodeError, ValueError):
+            return []
+    
+    def set_clauses(self, clauses: List[Clause]) -> None:
+        """Set clauses as JSON."""
+        self.clauses = json.dumps([clause.model_dump() for clause in clauses])
+    
+    def get_parties(self) -> List[Dict[str, Any]]:
+        """Get parsed parties from JSON."""
+        try:
+            return json.loads(self.parties)
+        except json.JSONDecodeError:
+            return []
+    
+    def set_parties(self, parties: List[Dict[str, Any]]) -> None:
+        """Set parties as JSON."""
+        self.parties = json.dumps(parties)
+    
+    def get_signatures(self) -> List[Signature]:
+        """Get parsed signatures from JSON."""
+        try:
+            signatures_data = json.loads(self.signatures)
+            return [Signature(**sig) for sig in signatures_data]
+        except (json.JSONDecodeError, ValueError):
+            return []
+    
+    def set_signatures(self, signatures: List[Signature]) -> None:
+        """Set signatures as JSON."""
+        self.signatures = json.dumps([sig.model_dump() for sig in signatures])
+    
+    def get_hipaa_entities(self) -> List[Dict[str, Any]]:
+        """Get parsed HIPAA entities from JSON."""
+        try:
+            return json.loads(self.hipaa_entities)
+        except json.JSONDecodeError:
+            return []
+    
+    def set_hipaa_entities(self, entities: List[Dict[str, Any]]) -> None:
+        """Set HIPAA entities as JSON."""
+        self.hipaa_entities = json.dumps(entities)
+    
+    def get_metadata(self) -> Dict[str, Any]:
+        """Get parsed metadata from JSON."""
+        try:
+            return json.loads(self.contract_metadata)
+        except json.JSONDecodeError:
+            return {}
+    
+    def set_metadata(self, metadata: Dict[str, Any]) -> None:
+        """Set metadata as JSON."""
+        self.contract_metadata = json.dumps(metadata)
+    
+    def get_content_hash(self) -> str:
+        """Get SHA-256 hash of contract content for signing."""
+        content = {
+            "id": str(self.id),
+            "title": self.title,
+            "description": self.description,
+            "clauses": self.clauses,
+            "parties": self.parties,
+            "version": self.version
+        }
+        content_str = json.dumps(content, sort_keys=True)
+        return hashlib.sha256(content_str.encode()).hexdigest()
+    
+    def can_transition_to(self, new_state: ContractState) -> bool:
+        """Check if contract can transition to new state."""
+        valid_transitions = {
+            ContractState.DRAFT: [ContractState.PROPOSED, ContractState.REVOKED],
+            ContractState.PROPOSED: [ContractState.SIGNED, ContractState.REVOKED, ContractState.DRAFT],
+            ContractState.SIGNED: [ContractState.REVOKED],
+            ContractState.REVOKED: [],  # Terminal state
+            ContractState.EXPIRED: []   # Terminal state
+        }
+        return new_state in valid_transitions.get(self.state, [])
+    
+    def is_fully_signed(self) -> bool:
+        """Check if contract is fully signed by all required parties."""
+        parties = self.get_parties()
+        signatures = self.get_signatures()
+        
+        # Check if all parties have signed
+        signed_party_ids = {sig.signer_id for sig in signatures}
+        required_party_ids = {party["id"] for party in parties}
+        
+        return required_party_ids.issubset(signed_party_ids)
+    
+    def get_unsigned_parties(self) -> List[Dict[str, Any]]:
+        """Get parties that haven't signed yet."""
+        parties = self.get_parties()
+        signatures = self.get_signatures()
+        signed_party_ids = {sig.signer_id for sig in signatures}
+        
+        return [party for party in parties if party["id"] not in signed_party_ids]
+
+
+class ContractCreateRequest(BaseModel):
+    """Request model for creating a contract."""
+    
+    title: str = Field(..., description="Contract title")
+    description: str = Field(..., description="Contract description")
+    clauses: List[Clause] = Field(..., description="Contract clauses")
+    parties: List[Dict[str, Any]] = Field(..., description="Contract parties")
+    is_hipaa_compliant: bool = Field(default=False, description="HIPAA compliance flag")
+    hipaa_entities: Optional[List[Dict[str, Any]]] = Field(default=None, description="HIPAA entities")
+    expires_at: Optional[datetime] = Field(default=None, description="Contract expiration")
+    metadata: Dict[str, Any] = Field(default_factory=dict, description="Additional metadata")
+    version: str = Field(default="1.0.0", description="Contract version")
+
+
+class ContractProposeRequest(BaseModel):
+    """Request model for proposing a contract."""
+    
+    proposed_to: List[str] = Field(..., description="IDs of parties to propose to")
+    message: Optional[str] = Field(default=None, description="Proposal message")
+
+
+class ContractSignRequest(BaseModel):
+    """Request model for signing a contract."""
+    
+    signer_id: str = Field(..., description="ID of the signing party")
+    signer_type: str = Field(..., description="Type of signer")
+    public_key: str = Field(..., description="Base64-encoded public key")
+    signature: str = Field(..., description="Base64-encoded Ed25519 signature")
+    metadata: Dict[str, Any] = Field(default_factory=dict, description="Signature metadata")
+
+
+class ContractRevokeRequest(BaseModel):
+    """Request model for revoking a contract."""
+    
+    reason: str = Field(..., description="Reason for revocation")
+    revoked_by: str = Field(..., description="ID of the revoking party")
+    metadata: Dict[str, Any] = Field(default_factory=dict, description="Revocation metadata")
+
+
+class ContractResponse(BaseModel):
+    """Response model for contract operations."""
+    
+    id: str
+    title: str
+    description: str
+    clauses: List[Clause]
+    parties: List[Dict[str, Any]]
+    state: ContractState
+    created_at: datetime
+    proposed_at: Optional[datetime]
+    signed_at: Optional[datetime]
+    revoked_at: Optional[datetime]
+    expires_at: Optional[datetime]
+    signatures: List[Dict[str, Any]]
+    is_hipaa_compliant: bool
+    hipaa_entities: List[Dict[str, Any]]
+    contract_metadata: Dict[str, Any]
+    version: str
+    created_by: str
+    last_modified: datetime
+    content_hash: str
+    is_fully_signed: bool
+    unsigned_parties: List[Dict[str, Any]]
+    
+    def model_dump(self, **kwargs):
+        """Override model_dump to handle datetime serialization."""
+        data = super().model_dump(**kwargs)
+        # Convert datetime objects to ISO format strings
+        for key, value in data.items():
+            if isinstance(value, datetime):
+                data[key] = value.isoformat()
+            elif key == "signatures" and isinstance(value, list):
+                # Handle nested Signature objects
+                data[key] = [sig.model_dump() if hasattr(sig, 'model_dump') else sig for sig in value]
+        return data
+    
+    @classmethod
+    def from_contract(cls, contract: Contract) -> "ContractResponse":
+        """Create response from contract model."""
+        # Pre-serialize signatures to handle datetime fields
+        signatures = contract.get_signatures()
+        serialized_signatures = [sig.model_dump() for sig in signatures]
+        
+        return cls(
+            id=str(contract.id),  # Convert UUID to string
+            title=contract.title,
+            description=contract.description,
+            clauses=contract.get_clauses(),
+            parties=contract.get_parties(),
+            state=contract.state,
+            created_at=contract.created_at,
+            proposed_at=contract.proposed_at,
+            signed_at=contract.signed_at,
+            revoked_at=contract.revoked_at,
+            expires_at=contract.expires_at,
+            signatures=serialized_signatures,  # Use pre-serialized signatures
+            is_hipaa_compliant=contract.is_hipaa_compliant,
+            hipaa_entities=contract.get_hipaa_entities(),
+            contract_metadata=contract.get_metadata(),
+            version=contract.version,
+            created_by=contract.created_by,
+            last_modified=contract.last_modified,
+            content_hash=contract.get_content_hash(),
+            is_fully_signed=contract.is_fully_signed(),
+            unsigned_parties=contract.get_unsigned_parties()
+        )

src/fastmcp/contracts/crypto.py

@@ -0,0 +1,230 @@
+"""Cryptographic utilities for contract signing and verification."""
+
+import base64
+from typing import Tuple
+
+import nacl.encoding
+import nacl.signing
+from nacl.exceptions import BadSignatureError
+
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class CryptoError(Exception):
+    """Cryptographic operation error."""
+    pass
+
+
+class Ed25519Signer:
+    """Ed25519 signature operations for contract signing."""
+    
+    def __init__(self, private_key: bytes = None):
+        """Initialize signer with optional private key.
+        
+        Args:
+            private_key: Optional private key bytes. If None, generates new key pair.
+        """
+        if private_key is None:
+            self._signing_key = nacl.signing.SigningKey.generate()
+        else:
+            self._signing_key = nacl.signing.SigningKey(private_key)
+        
+        self._verify_key = self._signing_key.verify_key
+    
+    @classmethod
+    def from_private_key_b64(cls, private_key_b64: str) -> "Ed25519Signer":
+        """Create signer from base64-encoded private key.
+        
+        Args:
+            private_key_b64: Base64-encoded private key
+            
+        Returns:
+            Ed25519Signer instance
+        """
+        try:
+            private_key = base64.b64decode(private_key_b64)
+            return cls(private_key)
+        except Exception as e:
+            raise CryptoError(f"Invalid private key format: {e}")
+    
+    @classmethod
+    def from_public_key_b64(cls, public_key_b64: str) -> "Ed25519Signer":
+        """Create signer from base64-encoded public key (verification only).
+        
+        Args:
+            public_key_b64: Base64-encoded public key
+            
+        Returns:
+            Ed25519Signer instance (verification only)
+        """
+        try:
+            public_key = base64.b64decode(public_key_b64)
+            verify_key = nacl.signing.VerifyKey(public_key)
+            signer = cls.__new__(cls)
+            signer._signing_key = None  # No private key for verification only
+            signer._verify_key = verify_key
+            return signer
+        except Exception as e:
+            raise CryptoError(f"Invalid public key format: {e}")
+    
+    def sign(self, message: str) -> str:
+        """Sign a message.
+        
+        Args:
+            message: Message to sign
+            
+        Returns:
+            Base64-encoded signature
+            
+        Raises:
+            CryptoError: If signing fails
+        """
+        if self._signing_key is None:
+            raise CryptoError("Cannot sign: no private key available")
+        
+        try:
+            message_bytes = message.encode('utf-8')
+            signed = self._signing_key.sign(message_bytes)
+            signature = signed.signature
+            return base64.b64encode(signature).decode('ascii')
+        except Exception as e:
+            raise CryptoError(f"Signing failed: {e}")
+    
+    def verify(self, message: str, signature: str) -> bool:
+        """Verify a signature.
+        
+        Args:
+            message: Original message
+            signature: Base64-encoded signature to verify
+            
+        Returns:
+            True if signature is valid, False otherwise
+        """
+        try:
+            message_bytes = message.encode('utf-8')
+            signature_bytes = base64.b64decode(signature)
+            self._verify_key.verify(message_bytes, signature_bytes)
+            return True
+        except (BadSignatureError, Exception) as e:
+            logger.debug(f"Signature verification failed: {e}")
+            return False
+    
+    def get_public_key_b64(self) -> str:
+        """Get base64-encoded public key.
+        
+        Returns:
+            Base64-encoded public key
+        """
+        return base64.b64encode(self._verify_key.encode()).decode('ascii')
+    
+    def get_private_key_b64(self) -> str:
+        """Get base64-encoded private key.
+        
+        Returns:
+            Base64-encoded private key
+            
+        Raises:
+            CryptoError: If no private key available
+        """
+        if self._signing_key is None:
+            raise CryptoError("No private key available")
+        
+        return base64.b64encode(self._signing_key.encode()).decode('ascii')
+    
+    def get_key_pair_b64(self) -> Tuple[str, str]:
+        """Get both public and private keys as base64 strings.
+        
+        Returns:
+            Tuple of (public_key_b64, private_key_b64)
+            
+        Raises:
+            CryptoError: If no private key available
+        """
+        return self.get_public_key_b64(), self.get_private_key_b64()
+
+
+class ContractSigner:
+    """High-level contract signing operations."""
+    
+    def __init__(self, signer: Ed25519Signer):
+        """Initialize with Ed25519 signer.
+        
+        Args:
+            signer: Ed25519Signer instance
+        """
+        self._signer = signer
+    
+    def sign_contract(self, contract_id: str, content_hash: str, signer_id: str, signer_type: str) -> str:
+        """Sign a contract.
+        
+        Args:
+            contract_id: Contract ID
+            content_hash: SHA-256 hash of contract content
+            signer_id: ID of the signing party
+            signer_type: Type of signer (e.g., 'provider', 'payor', 'patient')
+            
+        Returns:
+            Base64-encoded signature
+            
+        Raises:
+            CryptoError: If signing fails
+        """
+        # Create signing message
+        signing_message = f"{contract_id}:{content_hash}:{signer_id}:{signer_type}"
+        return self._signer.sign(signing_message)
+    
+    def verify_contract_signature(self, contract_id: str, content_hash: str, signer_id: str, 
+                                signer_type: str, signature: str) -> bool:
+        """Verify a contract signature.
+        
+        Args:
+            contract_id: Contract ID
+            content_hash: SHA-256 hash of contract content
+            signer_id: ID of the signing party
+            signer_type: Type of signer
+            signature: Base64-encoded signature to verify
+            
+        Returns:
+            True if signature is valid, False otherwise
+        """
+        # Create signing message
+        signing_message = f"{contract_id}:{content_hash}:{signer_id}:{signer_type}"
+        return self._signer.verify(signing_message, signature)
+    
+    def get_public_key_b64(self) -> str:
+        """Get base64-encoded public key."""
+        return self._signer.get_public_key_b64()
+    
+    def get_private_key_b64(self) -> str:
+        """Get base64-encoded private key."""
+        return self._signer.get_private_key_b64()
+
+
+def generate_key_pair() -> Tuple[str, str]:
+    """Generate a new Ed25519 key pair.
+    
+    Returns:
+        Tuple of (public_key_b64, private_key_b64)
+    """
+    signer = Ed25519Signer()
+    return signer.get_key_pair_b64()
+
+
+def verify_signature(public_key_b64: str, message: str, signature: str) -> bool:
+    """Verify a signature with a public key.
+    
+    Args:
+        public_key_b64: Base64-encoded public key
+        message: Original message
+        signature: Base64-encoded signature
+        
+    Returns:
+        True if signature is valid, False otherwise
+    """
+    try:
+        signer = Ed25519Signer.from_public_key_b64(public_key_b64)
+        return signer.verify(message, signature)
+    except CryptoError:
+        return False

src/fastmcp/contracts/engine.py

@@ -0,0 +1,357 @@
+"""Contract engine for managing contract lifecycle and operations."""
+
+from datetime import datetime
+from typing import List, Optional
+from uuid import UUID
+
+from sqlmodel import Session, create_engine
+
+from fastmcp.utilities.logging import get_logger
+
+from .contract import Contract, ContractState, ContractCreateRequest, ContractProposeRequest, ContractSignRequest, ContractRevokeRequest
+from .crypto import ContractSigner, CryptoError
+from .registry import ContractRegistry
+
+logger = get_logger(__name__)
+
+
+class ContractEngine:
+    """Engine for managing contract lifecycle and operations."""
+    
+    def __init__(self, database_url: str = "sqlite:///contracts.db"):
+        """Initialize contract engine with database connection.
+        
+        Args:
+            database_url: Database connection URL
+        """
+        self.engine = create_engine(database_url, echo=False)
+        self._create_tables()
+    
+    def _create_tables(self):
+        """Create database tables."""
+        try:
+            Contract.metadata.create_all(self.engine)
+            logger.info("Contract database tables created/verified")
+        except Exception as e:
+            logger.error(f"Failed to create contract tables: {e}")
+            raise
+    
+    def get_session(self) -> Session:
+        """Get database session.
+        
+        Returns:
+            SQLModel database session
+        """
+        return Session(self.engine)
+    
+    def get_registry(self) -> ContractRegistry:
+        """Get contract registry.
+        
+        Returns:
+            Contract registry instance
+        """
+        return ContractRegistry(self.get_session())
+    
+    async def create_contract(self, request: ContractCreateRequest, created_by: str) -> Contract:
+        """Create a new contract.
+        
+        Args:
+            request: Contract creation request
+            created_by: ID of the creating party
+            
+        Returns:
+            Created contract instance
+            
+        Raises:
+            ValueError: If contract creation fails
+        """
+        try:
+            registry = self.get_registry()
+            
+            # Convert request to contract data
+            contract_data = {
+                "title": request.title,
+                "description": request.description,
+                "clauses": [clause.model_dump() for clause in request.clauses],
+                "parties": request.parties,
+                "is_hipaa_compliant": request.is_hipaa_compliant,
+                "hipaa_entities": request.hipaa_entities or [],
+                "expires_at": request.expires_at,
+                "metadata": request.metadata,
+                "version": request.version
+            }
+            
+            contract = registry.create_contract(contract_data, created_by)
+            logger.info(f"Created contract {contract.id} by {created_by}")
+            return contract
+            
+        except Exception as e:
+            logger.error(f"Failed to create contract: {e}")
+            raise
+    
+    async def get_contract(self, contract_id: UUID) -> Optional[Contract]:
+        """Get a contract by ID.
+        
+        Args:
+            contract_id: Contract UUID
+            
+        Returns:
+            Contract instance or None if not found
+        """
+        try:
+            registry = self.get_registry()
+            return registry.get_contract(contract_id)
+        except Exception as e:
+            logger.error(f"Failed to get contract {contract_id}: {e}")
+            return None
+    
+    async def list_contracts(self, state: Optional[ContractState] = None, 
+                           created_by: Optional[str] = None) -> List[Contract]:
+        """List contracts with optional filtering.
+        
+        Args:
+            state: Optional state filter
+            created_by: Optional creator filter
+            
+        Returns:
+            List of contract instances
+        """
+        try:
+            registry = self.get_registry()
+            return registry.list_contracts(state, created_by)
+        except Exception as e:
+            logger.error(f"Failed to list contracts: {e}")
+            return []
+    
+    async def propose_contract(self, contract_id: UUID, request: ContractProposeRequest, 
+                              proposed_by: str) -> Optional[Contract]:
+        """Propose a contract to parties.
+        
+        Args:
+            contract_id: Contract UUID
+            request: Proposal request
+            proposed_by: ID of the proposing party
+            
+        Returns:
+            Updated contract instance or None if not found
+            
+        Raises:
+            ValueError: If proposal fails
+        """
+        try:
+            registry = self.get_registry()
+            
+            # Validate contract exists and can be proposed
+            contract = registry.get_contract(contract_id)
+            if not contract:
+                raise ValueError("Contract not found")
+            
+            if contract.state != ContractState.DRAFT:
+                raise ValueError(f"Cannot propose contract in state {contract.state}")
+            
+            # Update metadata with proposal info
+            metadata = {
+                "proposal": {
+                    "proposed_to": request.proposed_to,
+                    "message": request.message,
+                    "proposed_by": proposed_by,
+                    "timestamp": datetime.utcnow().isoformat()
+                }
+            }
+            
+            # Update state to proposed
+            updated_contract = registry.update_contract_state(
+                contract_id, ContractState.PROPOSED, proposed_by, metadata
+            )
+            
+            logger.info(f"Proposed contract {contract_id} to {request.proposed_to} by {proposed_by}")
+            return updated_contract
+            
+        except Exception as e:
+            logger.error(f"Failed to propose contract {contract_id}: {e}")
+            raise
+    
+    async def sign_contract(self, contract_id: UUID, request: ContractSignRequest) -> Optional[Contract]:
+        """Sign a contract.
+        
+        Args:
+            contract_id: Contract UUID
+            request: Signing request
+            
+        Returns:
+            Updated contract instance or None if not found
+            
+        Raises:
+            ValueError: If signing fails
+        """
+        try:
+            registry = self.get_registry()
+            
+            # Validate contract exists and can be signed
+            contract = registry.get_contract(contract_id)
+            if not contract:
+                raise ValueError("Contract not found")
+            
+            if contract.state not in [ContractState.PROPOSED, ContractState.SIGNED]:
+                raise ValueError(f"Cannot sign contract in state {contract.state}")
+            
+            # Check if party is already signed
+            existing_signatures = contract.get_signatures()
+            if any(sig.signer_id == request.signer_id for sig in existing_signatures):
+                raise ValueError("Party has already signed this contract")
+            
+            # Verify signature
+            if not self._verify_contract_signature(contract, request):
+                raise ValueError("Invalid signature")
+            
+            # Create signature object
+            from .contract import Signature
+            signature = Signature(
+                signer_id=request.signer_id,
+                signer_type=request.signer_type,
+                signature=request.signature,
+                public_key=request.public_key,
+                metadata=request.metadata
+            )
+            
+            # Add signature
+            updated_contract = registry.add_signature(contract_id, signature.model_dump())
+            
+            logger.info(f"Signed contract {contract_id} by {request.signer_id}")
+            return updated_contract
+            
+        except Exception as e:
+            logger.error(f"Failed to sign contract {contract_id}: {e}")
+            raise
+    
+    async def revoke_contract(self, contract_id: UUID, request: ContractRevokeRequest) -> Optional[Contract]:
+        """Revoke a contract.
+        
+        Args:
+            contract_id: Contract UUID
+            request: Revocation request
+            
+        Returns:
+            Updated contract instance or None if not found
+            
+        Raises:
+            ValueError: If revocation fails
+        """
+        try:
+            registry = self.get_registry()
+            
+            # Validate contract exists and can be revoked
+            contract = registry.get_contract(contract_id)
+            if not contract:
+                raise ValueError("Contract not found")
+            
+            if contract.state == ContractState.REVOKED:
+                raise ValueError("Contract is already revoked")
+            
+            if contract.state == ContractState.EXPIRED:
+                raise ValueError("Cannot revoke expired contract")
+            
+            # Revoke contract
+            updated_contract = registry.revoke_contract(
+                contract_id, request.reason, request.revoked_by, request.metadata
+            )
+            
+            logger.info(f"Revoked contract {contract_id} by {request.revoked_by}: {request.reason}")
+            return updated_contract
+            
+        except Exception as e:
+            logger.error(f"Failed to revoke contract {contract_id}: {e}")
+            raise
+    
+    def _verify_contract_signature(self, contract: Contract, request: ContractSignRequest) -> bool:
+        """Verify a contract signature.
+        
+        Args:
+            contract: Contract instance
+            request: Signing request
+            
+        Returns:
+            True if signature is valid, False otherwise
+        """
+        try:
+            from .crypto import verify_signature
+            
+            # Create signing message
+            signing_message = f"{contract.id}:{contract.get_content_hash()}:{request.signer_id}:{request.signer_type}"
+            
+            # Verify signature
+            return verify_signature(
+                request.public_key,
+                signing_message,
+                request.signature
+            )
+            
+        except Exception as e:
+            logger.error(f"Signature verification failed: {e}")
+            return False
+    
+    async def get_contracts_by_party(self, party_id: str) -> List[Contract]:
+        """Get contracts involving a specific party.
+        
+        Args:
+            party_id: Party ID to search for
+            
+        Returns:
+            List of contracts involving the party
+        """
+        try:
+            registry = self.get_registry()
+            return registry.get_contracts_by_party(party_id)
+        except Exception as e:
+            logger.error(f"Failed to get contracts for party {party_id}: {e}")
+            return []
+    
+    async def cleanup_expired_contracts(self) -> int:
+        """Mark expired contracts as expired.
+        
+        Returns:
+            Number of contracts marked as expired
+        """
+        try:
+            registry = self.get_registry()
+            count = registry.mark_contracts_expired()
+            logger.info(f"Marked {count} contracts as expired")
+            return count
+        except Exception as e:
+            logger.error(f"Failed to cleanup expired contracts: {e}")
+            return 0
+    
+    async def get_contract_statistics(self) -> dict:
+        """Get contract statistics.
+        
+        Returns:
+            Dictionary with contract statistics
+        """
+        try:
+            registry = self.get_registry()
+            
+            # Get all contracts
+            all_contracts = registry.list_contracts()
+            
+            # Count by state
+            state_counts = {}
+            for state in ContractState:
+                state_counts[state.value] = len([c for c in all_contracts if c.state == state])
+            
+            # Count HIPAA contracts
+            hipaa_count = len([c for c in all_contracts if c.is_hipaa_compliant])
+            
+            # Count signed contracts
+            signed_count = len([c for c in all_contracts if c.state == ContractState.SIGNED])
+            
+            return {
+                "total_contracts": len(all_contracts),
+                "by_state": state_counts,
+                "hipaa_compliant": hipaa_count,
+                "signed_contracts": signed_count,
+                "expired_contracts": len(registry.get_expired_contracts())
+            }
+            
+        except Exception as e:
+            logger.error(f"Failed to get contract statistics: {e}")
+            return {}

src/fastmcp/contracts/registry.py

@@ -0,0 +1,391 @@
+"""Contract registry for managing contract persistence and lifecycle."""
+
+import json
+from datetime import datetime
+from typing import List, Optional
+from uuid import UUID
+
+from sqlmodel import Session, select
+
+from fastmcp.utilities.logging import get_logger
+
+from .contract import Contract, ContractState, Signature
+from .crypto import ContractSigner, CryptoError
+
+logger = get_logger(__name__)
+
+
+class ContractRegistry:
+    """Registry for managing contract persistence and lifecycle operations."""
+    
+    def __init__(self, session: Session):
+        """Initialize registry with database session.
+        
+        Args:
+            session: SQLModel database session
+        """
+        self.session = session
+    
+    def create_contract(self, contract_data: dict, created_by: str) -> Contract:
+        """Create a new contract.
+        
+        Args:
+            contract_data: Contract data dictionary
+            created_by: ID of the creating party
+            
+        Returns:
+            Created contract instance
+            
+        Raises:
+            ValueError: If contract data is invalid
+        """
+        try:
+            # Create contract instance
+            contract = Contract(
+                title=contract_data["title"],
+                description=contract_data["description"],
+                created_by=created_by,
+                is_hipaa_compliant=contract_data.get("is_hipaa_compliant", False),
+                version=contract_data.get("version", "1.0.0")
+            )
+            
+            # Set clauses
+            if "clauses" in contract_data:
+                from .contract import Clause
+                clauses = [Clause(**clause) for clause in contract_data["clauses"]]
+                contract.set_clauses(clauses)
+            
+            # Set parties
+            if "parties" in contract_data:
+                contract.set_parties(contract_data["parties"])
+            
+            # Set HIPAA entities
+            if "hipaa_entities" in contract_data and contract_data["hipaa_entities"]:
+                contract.set_hipaa_entities(contract_data["hipaa_entities"])
+            
+            # Set metadata
+            if "metadata" in contract_data:
+                contract.set_metadata(contract_data["metadata"])
+            
+            # Set expiration
+            if "expires_at" in contract_data and contract_data["expires_at"]:
+                contract.expires_at = contract_data["expires_at"]
+            
+            # Save to database
+            self.session.add(contract)
+            self.session.commit()
+            self.session.refresh(contract)
+            
+            logger.info(f"Created contract {contract.id} by {created_by}")
+            return contract
+            
+        except Exception as e:
+            self.session.rollback()
+            logger.error(f"Failed to create contract: {e}")
+            raise ValueError(f"Contract creation failed: {e}")
+    
+    def get_contract(self, contract_id: UUID) -> Optional[Contract]:
+        """Get a contract by ID.
+        
+        Args:
+            contract_id: Contract UUID
+            
+        Returns:
+            Contract instance or None if not found
+        """
+        try:
+            statement = select(Contract).where(Contract.id == contract_id)
+            return self.session.exec(statement).first()
+        except Exception as e:
+            logger.error(f"Failed to get contract {contract_id}: {e}")
+            return None
+    
+    def list_contracts(self, state: Optional[ContractState] = None, 
+                      created_by: Optional[str] = None) -> List[Contract]:
+        """List contracts with optional filtering.
+        
+        Args:
+            state: Optional state filter
+            created_by: Optional creator filter
+            
+        Returns:
+            List of contract instances
+        """
+        try:
+            statement = select(Contract)
+            
+            if state is not None:
+                statement = statement.where(Contract.state == state)
+            
+            if created_by is not None:
+                statement = statement.where(Contract.created_by == created_by)
+            
+            statement = statement.order_by(Contract.created_at.desc())
+            return list(self.session.exec(statement))
+            
+        except Exception as e:
+            logger.error(f"Failed to list contracts: {e}")
+            return []
+    
+    def update_contract_state(self, contract_id: UUID, new_state: ContractState, 
+                            updated_by: str, metadata: Optional[dict] = None) -> Optional[Contract]:
+        """Update contract state with lifecycle validation.
+        
+        Args:
+            contract_id: Contract UUID
+            new_state: New state to transition to
+            updated_by: ID of the updating party
+            metadata: Optional metadata for the state change
+            
+        Returns:
+            Updated contract instance or None if not found
+            
+        Raises:
+            ValueError: If state transition is invalid
+        """
+        try:
+            contract = self.get_contract(contract_id)
+            if not contract:
+                return None
+            
+            # Validate state transition
+            if not contract.can_transition_to(new_state):
+                raise ValueError(f"Invalid state transition from {contract.state} to {new_state}")
+            
+            # Update state and timestamps
+            old_state = contract.state
+            contract.state = new_state
+            contract.last_modified = datetime.utcnow()
+            
+            # Set state-specific timestamps
+            if new_state == ContractState.PROPOSED and contract.proposed_at is None:
+                contract.proposed_at = datetime.utcnow()
+            elif new_state == ContractState.SIGNED and contract.signed_at is None:
+                contract.signed_at = datetime.utcnow()
+            elif new_state == ContractState.REVOKED and contract.revoked_at is None:
+                contract.revoked_at = datetime.utcnow()
+            
+            # Update metadata if provided
+            if metadata:
+                current_metadata = contract.get_metadata()
+                current_metadata.update(metadata)
+                current_metadata[f"state_change_{new_state}"] = {
+                    "timestamp": datetime.utcnow().isoformat(),
+                    "updated_by": updated_by,
+                    "previous_state": old_state
+                }
+                contract.set_metadata(current_metadata)
+            
+            # Save changes
+            self.session.add(contract)
+            self.session.commit()
+            self.session.refresh(contract)
+            
+            logger.info(f"Updated contract {contract_id} state from {old_state} to {new_state} by {updated_by}")
+            return contract
+            
+        except Exception as e:
+            self.session.rollback()
+            logger.error(f"Failed to update contract {contract_id} state: {e}")
+            raise
+    
+    def add_signature(self, contract_id: UUID, signature_data: dict) -> Optional[Contract]:
+        """Add a signature to a contract.
+        
+        Args:
+            contract_id: Contract UUID
+            signature_data: Signature data dictionary
+            
+        Returns:
+            Updated contract instance or None if not found
+            
+        Raises:
+            ValueError: If signature is invalid or contract not in signable state
+        """
+        try:
+            contract = self.get_contract(contract_id)
+            if not contract:
+                return None
+            
+            # Validate contract state
+            if contract.state not in [ContractState.PROPOSED, ContractState.SIGNED]:
+                raise ValueError(f"Cannot sign contract in state {contract.state}")
+            
+            # Create signature
+            from .contract import Signature
+            signature = Signature(**signature_data)
+            
+            # Verify signature
+            if not self._verify_signature(contract, signature):
+                raise ValueError("Invalid signature")
+            
+            # Add signature
+            signatures = contract.get_signatures()
+            signatures.append(signature)
+            contract.set_signatures(signatures)
+            
+            # Update state if fully signed
+            if contract.is_fully_signed() and contract.state == ContractState.PROPOSED:
+                # Use the proper state transition method
+                contract = self.update_contract_state(contract_id, ContractState.SIGNED, signature.signer_id)
+            
+            contract.last_modified = datetime.utcnow()
+            
+            # Save changes
+            self.session.add(contract)
+            self.session.commit()
+            self.session.refresh(contract)
+            
+            logger.info(f"Added signature to contract {contract_id} by {signature.signer_id}")
+            return contract
+            
+        except Exception as e:
+            self.session.rollback()
+            logger.error(f"Failed to add signature to contract {contract_id}: {e}")
+            raise
+    
+    def _verify_signature(self, contract: Contract, signature: Signature) -> bool:
+        """Verify a contract signature.
+        
+        Args:
+            contract: Contract instance
+            signature: Signature to verify
+            
+        Returns:
+            True if signature is valid, False otherwise
+        """
+        try:
+            from .crypto import verify_signature
+            
+            # Create signing message
+            signing_message = f"{contract.id}:{contract.get_content_hash()}:{signature.signer_id}:{signature.signer_type}"
+            
+            # Verify signature
+            return verify_signature(
+                signature.public_key,
+                signing_message,
+                signature.signature
+            )
+            
+        except Exception as e:
+            logger.error(f"Signature verification failed: {e}")
+            return False
+    
+    def revoke_contract(self, contract_id: UUID, reason: str, revoked_by: str, 
+                       metadata: Optional[dict] = None) -> Optional[Contract]:
+        """Revoke a contract.
+        
+        Args:
+            contract_id: Contract UUID
+            reason: Reason for revocation
+            revoked_by: ID of the revoking party
+            metadata: Optional revocation metadata
+            
+        Returns:
+            Updated contract instance or None if not found
+            
+        Raises:
+            ValueError: If contract cannot be revoked
+        """
+        try:
+            contract = self.get_contract(contract_id)
+            if not contract:
+                return None
+            
+            # Validate revocation
+            if contract.state == ContractState.REVOKED:
+                raise ValueError("Contract is already revoked")
+            
+            if contract.state == ContractState.EXPIRED:
+                raise ValueError("Cannot revoke expired contract")
+            
+            # Update metadata with revocation info
+            current_metadata = contract.get_metadata()
+            current_metadata["revocation"] = {
+                "reason": reason,
+                "revoked_by": revoked_by,
+                "timestamp": datetime.utcnow().isoformat()
+            }
+            if metadata:
+                current_metadata["revocation"].update(metadata)
+            
+            contract.set_metadata(current_metadata)
+            
+            # Update state
+            return self.update_contract_state(contract_id, ContractState.REVOKED, revoked_by)
+            
+        except Exception as e:
+            logger.error(f"Failed to revoke contract {contract_id}: {e}")
+            raise
+    
+    def get_contracts_by_party(self, party_id: str) -> List[Contract]:
+        """Get contracts involving a specific party.
+        
+        Args:
+            party_id: Party ID to search for
+            
+        Returns:
+            List of contracts involving the party
+        """
+        try:
+            statement = select(Contract)
+            contracts = list(self.session.exec(statement))
+            
+            # Filter contracts that involve the party
+            party_contracts = []
+            for contract in contracts:
+                parties = contract.get_parties()
+                if any(party.get("id") == party_id for party in parties):
+                    party_contracts.append(contract)
+            
+            return party_contracts
+            
+        except Exception as e:
+            logger.error(f"Failed to get contracts for party {party_id}: {e}")
+            return []
+    
+    def get_expired_contracts(self) -> List[Contract]:
+        """Get contracts that have expired.
+        
+        Returns:
+            List of expired contracts
+        """
+        try:
+            now = datetime.utcnow()
+            statement = select(Contract).where(
+                Contract.expires_at.isnot(None),
+                Contract.expires_at < now,
+                Contract.state.notin_([ContractState.EXPIRED, ContractState.REVOKED])
+            )
+            return list(self.session.exec(statement))
+            
+        except Exception as e:
+            logger.error(f"Failed to get expired contracts: {e}")
+            return []
+    
+    def mark_contracts_expired(self) -> int:
+        """Mark expired contracts as expired.
+        
+        Returns:
+            Number of contracts marked as expired
+        """
+        try:
+            expired_contracts = self.get_expired_contracts()
+            count = 0
+            
+            for contract in expired_contracts:
+                contract.state = ContractState.EXPIRED
+                contract.last_modified = datetime.utcnow()
+                self.session.add(contract)
+                count += 1
+            
+            if count > 0:
+                self.session.commit()
+                logger.info(f"Marked {count} contracts as expired")
+            
+            return count
+            
+        except Exception as e:
+            self.session.rollback()
+            logger.error(f"Failed to mark contracts as expired: {e}")
+            return 0

src/fastmcp/server/contract_routes.py

@@ -0,0 +1,425 @@
+"""Contract 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.contracts import ContractEngine, ContractState
+from fastmcp.contracts.contract import (
+    ContractCreateRequest, ContractProposeRequest, ContractSignRequest, 
+    ContractRevokeRequest, ContractResponse
+)
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+async def create_contract_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for creating contracts.
+    
+    Expected JSON body:
+    {
+        "title": "Contract Title",
+        "description": "Contract Description",
+        "clauses": [...],
+        "parties": [...],
+        "is_hipaa_compliant": false,
+        "expires_at": "2024-12-31T23:59:59Z"
+    }
+    """
+    try:
+        # Parse request body
+        body = await request.json()
+        
+        # Get contract engine from request state
+        contract_engine: ContractEngine = request.app.state.contract_engine
+        
+        # Get creator from headers or body
+        created_by = request.headers.get("X-User-ID") or body.get("created_by", "anonymous")
+        
+        # Create contract request
+        contract_request = ContractCreateRequest(**body)
+        
+        # Create contract
+        contract = await contract_engine.create_contract(contract_request, created_by)
+        
+        # Return contract response
+        response = ContractResponse.from_contract(contract)
+        return JSONResponse(
+            status_code=201,
+            content=response.model_dump()
+        )
+        
+    except Exception as e:
+        logger.error(f"Contract creation error: {e}")
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Contract creation failed",
+                "reason": str(e)
+            }
+        )
+
+
+async def get_contract_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for getting a contract by ID."""
+    try:
+        # Get contract ID from path
+        contract_id = UUID(request.path_params["id"])
+        
+        # Get contract engine from request state
+        contract_engine: ContractEngine = request.app.state.contract_engine
+        
+        # Get contract
+        contract = await contract_engine.get_contract(contract_id)
+        
+        if not contract:
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "error": "Contract not found",
+                    "contract_id": str(contract_id)
+                }
+            )
+        
+        # Return contract response
+        response = ContractResponse.from_contract(contract)
+        return JSONResponse(
+            status_code=200,
+            content=response.model_dump()
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid contract ID",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Contract retrieval error: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Contract retrieval failed",
+                "reason": str(e)
+            }
+        )
+
+
+async def list_contracts_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for listing contracts."""
+    try:
+        # Get query parameters
+        state = request.query_params.get("state")
+        created_by = request.query_params.get("created_by")
+        
+        # Get contract engine from request state
+        contract_engine: ContractEngine = request.app.state.contract_engine
+        
+        # Parse state if provided
+        contract_state = None
+        if state:
+            try:
+                contract_state = ContractState(state)
+            except ValueError:
+                return JSONResponse(
+                    status_code=400,
+                    content={
+                        "error": "Invalid state",
+                        "valid_states": [s.value for s in ContractState]
+                    }
+                )
+        
+        # List contracts
+        contracts = await contract_engine.list_contracts(contract_state, created_by)
+        
+        # Convert to response format
+        responses = [ContractResponse.from_contract(contract).model_dump() for contract in contracts]
+        
+        return JSONResponse(
+            status_code=200,
+            content={
+                "contracts": responses,
+                "count": len(responses)
+            }
+        )
+        
+    except Exception as e:
+        logger.error(f"Contract listing error: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Contract listing failed",
+                "reason": str(e)
+            }
+        )
+
+
+async def propose_contract_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for proposing contracts.
+    
+    Expected JSON body:
+    {
+        "proposed_to": ["party1", "party2"],
+        "message": "Please review and sign this contract"
+    }
+    """
+    try:
+        # Get contract ID from path
+        contract_id = UUID(request.path_params["id"])
+        
+        # Parse request body
+        body = await request.json()
+        
+        # Get contract engine from request state
+        contract_engine: ContractEngine = request.app.state.contract_engine
+        
+        # Get proposer from headers or body
+        proposed_by = request.headers.get("X-User-ID") or body.get("proposed_by", "anonymous")
+        
+        # Create proposal request
+        proposal_request = ContractProposeRequest(**body)
+        
+        # Propose contract
+        contract = await contract_engine.propose_contract(contract_id, proposal_request, proposed_by)
+        
+        if not contract:
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "error": "Contract not found",
+                    "contract_id": str(contract_id)
+                }
+            )
+        
+        # Return contract response
+        response = ContractResponse.from_contract(contract)
+        return JSONResponse(
+            status_code=200,
+            content=response.model_dump()
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid request",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Contract proposal error: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Contract proposal failed",
+                "reason": str(e)
+            }
+        )
+
+
+async def sign_contract_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for signing contracts.
+    
+    Expected JSON body:
+    {
+        "signer_id": "party1",
+        "signer_type": "provider",
+        "public_key": "base64_public_key",
+        "signature": "base64_signature"
+    }
+    """
+    try:
+        # Get contract ID from path
+        contract_id = UUID(request.path_params["id"])
+        
+        # Parse request body
+        body = await request.json()
+        
+        # Get contract engine from request state
+        contract_engine: ContractEngine = request.app.state.contract_engine
+        
+        # Create signing request
+        sign_request = ContractSignRequest(**body)
+        
+        # Sign contract
+        contract = await contract_engine.sign_contract(contract_id, sign_request)
+        
+        if not contract:
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "error": "Contract not found",
+                    "contract_id": str(contract_id)
+                }
+            )
+        
+        # Return contract response
+        response = ContractResponse.from_contract(contract)
+        return JSONResponse(
+            status_code=200,
+            content=response.model_dump()
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid request",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Contract signing error: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Contract signing failed",
+                "reason": str(e)
+            }
+        )
+
+
+async def revoke_contract_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for revoking contracts.
+    
+    Expected JSON body:
+    {
+        "reason": "Contract terms violated",
+        "revoked_by": "party1"
+    }
+    """
+    try:
+        # Get contract ID from path
+        contract_id = UUID(request.path_params["id"])
+        
+        # Parse request body
+        body = await request.json()
+        
+        # Get contract engine from request state
+        contract_engine: ContractEngine = request.app.state.contract_engine
+        
+        # Create revocation request
+        revoke_request = ContractRevokeRequest(**body)
+        
+        # Revoke contract
+        contract = await contract_engine.revoke_contract(contract_id, revoke_request)
+        
+        if not contract:
+            return JSONResponse(
+                status_code=404,
+                content={
+                    "error": "Contract not found",
+                    "contract_id": str(contract_id)
+                }
+            )
+        
+        # Return contract response
+        response = ContractResponse.from_contract(contract)
+        return JSONResponse(
+            status_code=200,
+            content=response.model_dump()
+        )
+        
+    except ValueError as e:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Invalid request",
+                "reason": str(e)
+            }
+        )
+    except Exception as e:
+        logger.error(f"Contract revocation error: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Contract revocation failed",
+                "reason": str(e)
+            }
+        )
+
+
+async def get_contract_statistics_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for getting contract statistics."""
+    try:
+        # Get contract engine from request state
+        contract_engine: ContractEngine = request.app.state.contract_engine
+        
+        # Get statistics
+        stats = await contract_engine.get_contract_statistics()
+        
+        return JSONResponse(
+            status_code=200,
+            content=stats
+        )
+        
+    except Exception as e:
+        logger.error(f"Contract statistics error: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Contract statistics failed",
+                "reason": str(e)
+            }
+        )
+
+
+def create_contract_routes(contract_engine: ContractEngine) -> List[Route]:
+    """Create contract management routes.
+    
+    Args:
+        contract_engine: The contract engine instance
+        
+    Returns:
+        List of Starlette Route objects for contract management
+    """
+    def endpoint_with_engine(endpoint_func):
+        async def wrapper(request: Request) -> JSONResponse:
+            # Store contract engine in app state for access in endpoint
+            request.app.state.contract_engine = contract_engine
+            return await endpoint_func(request)
+        return wrapper
+    
+    return [
+        Route(
+            path="/contracts",
+            endpoint=endpoint_with_engine(create_contract_endpoint),
+            methods=["POST"]
+        ),
+        Route(
+            path="/contracts",
+            endpoint=endpoint_with_engine(list_contracts_endpoint),
+            methods=["GET"]
+        ),
+        Route(
+            path="/contracts/statistics",
+            endpoint=endpoint_with_engine(get_contract_statistics_endpoint),
+            methods=["GET"]
+        ),
+        Route(
+            path="/contracts/{id}",
+            endpoint=endpoint_with_engine(get_contract_endpoint),
+            methods=["GET"]
+        ),
+        Route(
+            path="/contracts/{id}/propose",
+            endpoint=endpoint_with_engine(propose_contract_endpoint),
+            methods=["POST"]
+        ),
+        Route(
+            path="/contracts/{id}/sign",
+            endpoint=endpoint_with_engine(sign_contract_endpoint),
+            methods=["POST"]
+        ),
+        Route(
+            path="/contracts/{id}/revoke",
+            endpoint=endpoint_with_engine(revoke_contract_endpoint),
+            methods=["POST"]
+        )
+    ]

src/fastmcp/server/server.py

@@ -65,6 +65,7 @@ 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
 from fastmcp.utilities.logging import get_logger
@@ -178,6 +179,7 @@ class FastMCP(Generic[LifespanResultT]):
         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,
             mask_error_details=mask_error_details,
@@ -522,6 +524,13 @@ class FastMCP(Generic[LifespanResultT]):
             
             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
+            
+            contract_routes = create_contract_routes(self._contract_engine)
+            routes.extend(contract_routes)
 
         # Recursively get routes from mounted servers
         for mounted_server in self._mounted_servers:
@@ -579,6 +588,31 @@ class FastMCP(Generic[LifespanResultT]):
         """
         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.
+        
+        Args:
+            contract_engine: Optional contract engine instance. If None, creates a new one.
+            database_url: Database URL for contract persistence
+            
+        Returns:
+            The contract engine instance
+        """
+        if contract_engine is None:
+            contract_engine = ContractEngine(database_url)
+        
+        self._contract_engine = contract_engine
+        logger.info("Contract engine enabled for server")
+        return contract_engine
+    
+    def get_contract_engine(self) -> Optional[ContractEngine]:
+        """Get the contract engine instance.
+        
+        Returns:
+            The contract engine instance, or None if not enabled
+        """
+        return self._contract_engine
+
     async def _mcp_list_tools(self) -> list[MCPTool]:
         logger.debug("Handler called: list_tools")
 

tests/contracts/__init__.py

@@ -0,0 +1 @@
+"""Tests for the contract management system."""

tests/contracts/test_contract_engine.py

@@ -0,0 +1,408 @@
+"""Tests for the contract engine."""
+
+import pytest
+from datetime import datetime, timedelta
+from uuid import UUID
+
+from fastmcp.contracts import ContractEngine, ContractState
+from fastmcp.contracts.contract import (
+    Contract, Clause, Signature, ContractCreateRequest, ContractProposeRequest,
+    ContractSignRequest, ContractRevokeRequest
+)
+from fastmcp.contracts.crypto import Ed25519Signer, generate_key_pair
+
+
+class TestContractEngine:
+    """Test the contract engine functionality."""
+
+    @pytest.fixture
+    def contract_engine(self):
+        """Create a contract engine for testing."""
+        return ContractEngine("sqlite:///:memory:")
+
+    @pytest.fixture
+    def sample_clauses(self):
+        """Create sample clauses for testing."""
+        return [
+            Clause(
+                title="Data Handling",
+                content="All data must be handled in accordance with HIPAA regulations.",
+                type="hipaa"
+            ),
+            Clause(
+                title="Access Control",
+                content="Only authorized personnel may access patient data.",
+                type="security"
+            )
+        ]
+
+    @pytest.fixture
+    def sample_parties(self):
+        """Create sample parties for testing."""
+        return [
+            {
+                "id": "provider1",
+                "name": "Healthcare Provider",
+                "type": "provider",
+                "email": "provider@example.com"
+            },
+            {
+                "id": "patient1",
+                "name": "John Doe",
+                "type": "patient",
+                "email": "patient@example.com"
+            }
+        ]
+
+    @pytest.fixture
+    def sample_contract_request(self, sample_clauses, sample_parties):
+        """Create a sample contract creation request."""
+        return ContractCreateRequest(
+            title="HIPAA Data Sharing Agreement",
+            description="Agreement for sharing patient data between healthcare providers",
+            clauses=sample_clauses,
+            parties=sample_parties,
+            is_hipaa_compliant=True,
+            expires_at=datetime.utcnow() + timedelta(days=365)
+        )
+
+    def test_contract_engine_initialization(self, contract_engine):
+        """Test contract engine initialization."""
+        assert contract_engine.engine is not None
+        assert contract_engine.get_registry() is not None
+
+    @pytest.mark.asyncio
+    async def test_create_contract(self, contract_engine, sample_contract_request):
+        """Test contract creation."""
+        created_by = "admin"
+        
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        assert contract is not None
+        assert contract.title == sample_contract_request.title
+        assert contract.description == sample_contract_request.description
+        assert contract.state == ContractState.DRAFT
+        assert contract.created_by == created_by
+        assert contract.is_hipaa_compliant is True
+        assert len(contract.get_clauses()) == 2
+        assert len(contract.get_parties()) == 2
+
+    @pytest.mark.asyncio
+    async def test_get_contract(self, contract_engine, sample_contract_request):
+        """Test getting a contract by ID."""
+        created_by = "admin"
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        retrieved_contract = await contract_engine.get_contract(contract.id)
+        
+        assert retrieved_contract is not None
+        assert retrieved_contract.id == contract.id
+        assert retrieved_contract.title == contract.title
+
+    @pytest.mark.asyncio
+    async def test_get_contract_not_found(self, contract_engine):
+        """Test getting a non-existent contract."""
+        fake_id = UUID("12345678-1234-1234-1234-123456789012")
+        contract = await contract_engine.get_contract(fake_id)
+        
+        assert contract is None
+
+    @pytest.mark.asyncio
+    async def test_list_contracts(self, contract_engine, sample_contract_request):
+        """Test listing contracts."""
+        created_by = "admin"
+        
+        # Create multiple contracts
+        contract1 = await contract_engine.create_contract(sample_contract_request, created_by)
+        contract2 = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        # List all contracts
+        all_contracts = await contract_engine.list_contracts()
+        assert len(all_contracts) == 2
+        
+        # List contracts by state
+        draft_contracts = await contract_engine.list_contracts(state=ContractState.DRAFT)
+        assert len(draft_contracts) == 2
+        
+        # List contracts by creator
+        admin_contracts = await contract_engine.list_contracts(created_by=created_by)
+        assert len(admin_contracts) == 2
+
+    @pytest.mark.asyncio
+    async def test_propose_contract(self, contract_engine, sample_contract_request):
+        """Test proposing a contract."""
+        created_by = "admin"
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        proposal_request = ContractProposeRequest(
+            proposed_to=["provider1", "patient1"],
+            message="Please review and sign this contract"
+        )
+        
+        proposed_contract = await contract_engine.propose_contract(
+            contract.id, proposal_request, created_by
+        )
+        
+        assert proposed_contract is not None
+        assert proposed_contract.state == ContractState.PROPOSED
+        assert proposed_contract.proposed_at is not None
+
+    @pytest.mark.asyncio
+    async def test_propose_contract_invalid_state(self, contract_engine, sample_contract_request):
+        """Test proposing a contract in invalid state."""
+        created_by = "admin"
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        # First propose the contract
+        proposal_request = ContractProposeRequest(proposed_to=["provider1"])
+        await contract_engine.propose_contract(contract.id, proposal_request, created_by)
+        
+        # Try to propose again (should fail)
+        with pytest.raises(ValueError, match="Cannot propose contract in state ContractState.PROPOSED"):
+            await contract_engine.propose_contract(contract.id, proposal_request, created_by)
+
+    @pytest.mark.asyncio
+    async def test_sign_contract(self, contract_engine, sample_contract_request):
+        """Test signing a contract."""
+        created_by = "admin"
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        # Propose the contract
+        proposal_request = ContractProposeRequest(proposed_to=["provider1"])
+        await contract_engine.propose_contract(contract.id, proposal_request, created_by)
+        
+        # Generate key pair for signing
+        public_key, private_key = generate_key_pair()
+        signer = Ed25519Signer.from_private_key_b64(private_key)
+        
+        # Create signing message
+        signing_message = f"{contract.id}:{contract.get_content_hash()}:provider1:provider"
+        signature = signer.sign(signing_message)
+        
+        # Sign the contract
+        sign_request = ContractSignRequest(
+            signer_id="provider1",
+            signer_type="provider",
+            public_key=public_key,
+            signature=signature
+        )
+        
+        signed_contract = await contract_engine.sign_contract(contract.id, sign_request)
+        
+        assert signed_contract is not None
+        assert len(signed_contract.get_signatures()) == 1
+        assert signed_contract.get_signatures()[0].signer_id == "provider1"
+
+    @pytest.mark.asyncio
+    async def test_sign_contract_invalid_signature(self, contract_engine, sample_contract_request):
+        """Test signing a contract with invalid signature."""
+        created_by = "admin"
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        # Propose the contract
+        proposal_request = ContractProposeRequest(proposed_to=["provider1"])
+        await contract_engine.propose_contract(contract.id, proposal_request, created_by)
+        
+        # Generate key pair for signing
+        public_key, private_key = generate_key_pair()
+        signer = Ed25519Signer.from_private_key_b64(private_key)
+        
+        # Create invalid signing message (wrong content hash)
+        invalid_signing_message = f"{contract.id}:invalid_hash:provider1:provider"
+        signature = signer.sign(invalid_signing_message)
+        
+        # Try to sign with invalid signature
+        sign_request = ContractSignRequest(
+            signer_id="provider1",
+            signer_type="provider",
+            public_key=public_key,
+            signature=signature
+        )
+        
+        with pytest.raises(ValueError, match="Invalid signature"):
+            await contract_engine.sign_contract(contract.id, sign_request)
+
+    @pytest.mark.asyncio
+    async def test_revoke_contract(self, contract_engine, sample_contract_request):
+        """Test revoking a contract."""
+        created_by = "admin"
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        revoke_request = ContractRevokeRequest(
+            reason="Contract terms violated",
+            revoked_by="admin"
+        )
+        
+        revoked_contract = await contract_engine.revoke_contract(contract.id, revoke_request)
+        
+        assert revoked_contract is not None
+        assert revoked_contract.state == ContractState.REVOKED
+        assert revoked_contract.revoked_at is not None
+
+    @pytest.mark.asyncio
+    async def test_revoke_contract_already_revoked(self, contract_engine, sample_contract_request):
+        """Test revoking an already revoked contract."""
+        created_by = "admin"
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        # First revoke the contract
+        revoke_request = ContractRevokeRequest(
+            reason="Contract terms violated",
+            revoked_by="admin"
+        )
+        await contract_engine.revoke_contract(contract.id, revoke_request)
+        
+        # Try to revoke again (should fail)
+        with pytest.raises(ValueError, match="Contract is already revoked"):
+            await contract_engine.revoke_contract(contract.id, revoke_request)
+
+    @pytest.mark.asyncio
+    async def test_get_contracts_by_party(self, contract_engine, sample_contract_request):
+        """Test getting contracts by party."""
+        created_by = "admin"
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        # Get contracts for provider1
+        provider_contracts = await contract_engine.get_contracts_by_party("provider1")
+        assert len(provider_contracts) == 1
+        assert provider_contracts[0].id == contract.id
+        
+        # Get contracts for non-existent party
+        empty_contracts = await contract_engine.get_contracts_by_party("nonexistent")
+        assert len(empty_contracts) == 0
+
+    @pytest.mark.asyncio
+    async def test_cleanup_expired_contracts(self, contract_engine, sample_contract_request):
+        """Test cleanup of expired contracts."""
+        created_by = "admin"
+        
+        # Create contract with past expiration
+        past_expiration = datetime.utcnow() - timedelta(days=1)
+        sample_contract_request.expires_at = past_expiration
+        
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        # Cleanup expired contracts
+        count = await contract_engine.cleanup_expired_contracts()
+        
+        assert count == 1
+        
+        # Verify contract is marked as expired
+        expired_contract = await contract_engine.get_contract(contract.id)
+        assert expired_contract.state == ContractState.EXPIRED
+
+    @pytest.mark.asyncio
+    async def test_get_contract_statistics(self, contract_engine, sample_contract_request):
+        """Test getting contract statistics."""
+        created_by = "admin"
+        
+        # Create multiple contracts
+        contract1 = await contract_engine.create_contract(sample_contract_request, created_by)
+        contract2 = await contract_engine.create_contract(sample_contract_request, created_by)
+        
+        # Propose one contract
+        proposal_request = ContractProposeRequest(proposed_to=["provider1"])
+        await contract_engine.propose_contract(contract1.id, proposal_request, created_by)
+        
+        # Get statistics
+        stats = await contract_engine.get_contract_statistics()
+        
+        assert stats["total_contracts"] == 2
+        assert stats["by_state"]["draft"] == 1
+        assert stats["by_state"]["proposed"] == 1
+        assert stats["hipaa_compliant"] == 2
+        assert stats["signed_contracts"] == 0
+
+
+class TestContractLifecycle:
+    """Test complete contract lifecycle."""
+
+    @pytest.fixture
+    def contract_engine(self):
+        """Create a contract engine for testing."""
+        return ContractEngine("sqlite:///:memory:")
+
+    @pytest.fixture
+    def sample_contract_request(self):
+        """Create a sample contract creation request."""
+        return ContractCreateRequest(
+            title="Test Contract",
+            description="A test contract for lifecycle testing",
+            clauses=[
+                Clause(
+                    title="Test Clause",
+                    content="This is a test clause",
+                    type="test"
+                )
+            ],
+            parties=[
+                {
+                    "id": "party1",
+                    "name": "Test Party 1",
+                    "type": "provider"
+                },
+                {
+                    "id": "party2",
+                    "name": "Test Party 2",
+                    "type": "patient"
+                }
+            ]
+        )
+
+    @pytest.mark.asyncio
+    async def test_complete_contract_lifecycle(self, contract_engine, sample_contract_request):
+        """Test complete contract lifecycle: create → propose → sign → revoke."""
+        created_by = "admin"
+        
+        # 1. Create contract
+        contract = await contract_engine.create_contract(sample_contract_request, created_by)
+        assert contract.state == ContractState.DRAFT
+        
+        # 2. Propose contract
+        proposal_request = ContractProposeRequest(
+            proposed_to=["party1", "party2"],
+            message="Please review and sign"
+        )
+        contract = await contract_engine.propose_contract(contract.id, proposal_request, created_by)
+        assert contract.state == ContractState.PROPOSED
+        
+        # 3. Sign contract (party1)
+        public_key1, private_key1 = generate_key_pair()
+        signer1 = Ed25519Signer.from_private_key_b64(private_key1)
+        signing_message1 = f"{contract.id}:{contract.get_content_hash()}:party1:provider"
+        signature1 = signer1.sign(signing_message1)
+        
+        sign_request1 = ContractSignRequest(
+            signer_id="party1",
+            signer_type="provider",
+            public_key=public_key1,
+            signature=signature1
+        )
+        contract = await contract_engine.sign_contract(contract.id, sign_request1)
+        assert len(contract.get_signatures()) == 1
+        assert contract.state == ContractState.PROPOSED  # Still proposed until all parties sign
+        
+        # 4. Sign contract (party2)
+        public_key2, private_key2 = generate_key_pair()
+        signer2 = Ed25519Signer.from_private_key_b64(private_key2)
+        signing_message2 = f"{contract.id}:{contract.get_content_hash()}:party2:patient"
+        signature2 = signer2.sign(signing_message2)
+        
+        sign_request2 = ContractSignRequest(
+            signer_id="party2",
+            signer_type="patient",
+            public_key=public_key2,
+            signature=signature2
+        )
+        contract = await contract_engine.sign_contract(contract.id, sign_request2)
+        assert len(contract.get_signatures()) == 2
+        assert contract.state == ContractState.SIGNED  # Now fully signed
+        
+        # 5. Revoke contract
+        revoke_request = ContractRevokeRequest(
+            reason="Contract terms violated",
+            revoked_by="admin"
+        )
+        contract = await contract_engine.revoke_contract(contract.id, revoke_request)
+        assert contract.state == ContractState.REVOKED
+        assert contract.revoked_at is not None

tests/contracts/test_contract_http.py

@@ -0,0 +1,298 @@
+"""Tests for contract HTTP endpoints."""
+
+import pytest
+from datetime import datetime, timedelta
+from uuid import UUID
+
+from fastmcp import FastMCP
+from fastmcp.contracts import ContractEngine, ContractState
+from fastmcp.contracts.contract import (
+    Clause, ContractCreateRequest, ContractProposeRequest,
+    ContractSignRequest, ContractRevokeRequest
+)
+from fastmcp.contracts.crypto import generate_key_pair, Ed25519Signer
+from starlette.routing import Route
+
+
+class TestContractHTTPEndpoint:
+    """Test the contract management HTTP endpoints."""
+
+    @pytest.fixture
+    def server_with_contracts(self):
+        """Create a server with contract engine enabled."""
+        server = FastMCP("Test Contract Server")
+        # Use in-memory database for testing
+        contract_engine = server.enable_contract_engine(database_url="sqlite:///:memory:")
+        return server
+
+    @pytest.fixture
+    def app(self, server_with_contracts):
+        """Create the HTTP app with contract endpoints."""
+        return server_with_contracts.http_app(transport="sse")
+
+    def test_contract_endpoints_exist(self, app):
+        """Test that all contract endpoints exist in the app."""
+        expected_paths = [
+            "/contracts",
+            "/contracts/{id}",
+            "/contracts/{id}/propose",
+            "/contracts/{id}/sign",
+            "/contracts/{id}/revoke",
+            "/contracts/statistics"
+        ]
+        
+        # Check that all contract routes exist
+        contract_routes_found = set()
+        for route in app.routes:
+            if isinstance(route, Route):
+                if route.path in expected_paths:
+                    contract_routes_found.add(route.path)
+        
+        assert len(contract_routes_found) == len(expected_paths), f"Expected {len(expected_paths)} unique contract routes, found {len(contract_routes_found)}: {sorted(contract_routes_found)}"
+
+    def test_contract_engine_integration(self, server_with_contracts):
+        """Test that contract engine is properly integrated with the server."""
+        # Check that contract engine is enabled
+        assert server_with_contracts.get_contract_engine() is not None
+        
+        # Check that contract engine is the right type
+        contract_engine = server_with_contracts.get_contract_engine()
+        assert isinstance(contract_engine, ContractEngine)
+
+    @pytest.mark.asyncio
+    async def test_create_contract_endpoint(self, app):
+        """Test the create contract endpoint."""
+        import httpx
+        
+        contract_data = {
+            "title": "Test Contract",
+            "description": "A test contract",
+            "clauses": [
+                {
+                    "title": "Test Clause",
+                    "content": "This is a test clause",
+                    "type": "test"
+                }
+            ],
+            "parties": [
+                {
+                    "id": "party1",
+                    "name": "Test Party",
+                    "type": "provider"
+                }
+            ],
+            "is_hipaa_compliant": False
+        }
+        
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            response = await client.post("/contracts", json=contract_data)
+            
+            assert response.status_code == 201
+            data = response.json()
+            assert data["title"] == contract_data["title"]
+            assert data["description"] == contract_data["description"]
+            assert data["state"] == "draft"
+            assert data["is_hipaa_compliant"] is False
+
+    @pytest.mark.asyncio
+    async def test_get_contract_endpoint(self, app):
+        """Test the get contract endpoint."""
+        import httpx
+        
+        # First create a contract
+        contract_data = {
+            "title": "Test Contract",
+            "description": "A test contract",
+            "clauses": [],
+            "parties": []
+        }
+        
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            # Create contract
+            create_response = await client.post("/contracts", json=contract_data)
+            assert create_response.status_code == 201
+            contract_id = create_response.json()["id"]
+            
+            # Get contract
+            get_response = await client.get(f"/contracts/{contract_id}")
+            assert get_response.status_code == 200
+            data = get_response.json()
+            assert data["id"] == contract_id
+            assert data["title"] == contract_data["title"]
+
+    @pytest.mark.asyncio
+    async def test_get_contract_not_found(self, app):
+        """Test getting a non-existent contract."""
+        import httpx
+        
+        fake_id = "12345678-1234-1234-1234-123456789012"
+        
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            response = await client.get(f"/contracts/{fake_id}")
+            assert response.status_code == 404
+            data = response.json()
+            assert "not found" in data["error"].lower()
+
+    @pytest.mark.asyncio
+    async def test_list_contracts_endpoint(self, app):
+        """Test the list contracts endpoint."""
+        import httpx
+        
+        # Create multiple contracts
+        contract_data = {
+            "title": "Test Contract",
+            "description": "A test contract",
+            "clauses": [],
+            "parties": []
+        }
+        
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            # Create contracts
+            await client.post("/contracts", json=contract_data)
+            await client.post("/contracts", json=contract_data)
+            
+            # List contracts
+            response = await client.get("/contracts")
+            assert response.status_code == 200
+            data = response.json()
+            assert "contracts" in data
+            assert "count" in data
+            assert data["count"] == 2
+            assert len(data["contracts"]) == 2
+
+    @pytest.mark.asyncio
+    async def test_propose_contract_endpoint(self, app):
+        """Test the propose contract endpoint."""
+        import httpx
+        
+        # First create a contract
+        contract_data = {
+            "title": "Test Contract",
+            "description": "A test contract",
+            "clauses": [],
+            "parties": []
+        }
+        
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            # Create contract
+            create_response = await client.post("/contracts", json=contract_data)
+            contract_id = create_response.json()["id"]
+            
+            # Propose contract
+            proposal_data = {
+                "proposed_to": ["party1", "party2"],
+                "message": "Please review and sign"
+            }
+            
+            response = await client.post(f"/contracts/{contract_id}/propose", json=proposal_data)
+            assert response.status_code == 200
+            data = response.json()
+            assert data["state"] == "proposed"
+            assert data["proposed_at"] is not None
+
+    @pytest.mark.asyncio
+    async def test_sign_contract_endpoint(self, app):
+        """Test the sign contract endpoint."""
+        import httpx
+        
+        # First create and propose a contract
+        contract_data = {
+            "title": "Test Contract",
+            "description": "A test contract",
+            "clauses": [],
+            "parties": [{"id": "party1", "name": "Test Party", "type": "provider"}]
+        }
+        
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            # Create contract
+            create_response = await client.post("/contracts", json=contract_data)
+            contract_id = create_response.json()["id"]
+            
+            # Propose contract
+            proposal_data = {"proposed_to": ["party1"]}
+            await client.post(f"/contracts/{contract_id}/propose", json=proposal_data)
+            
+            # Generate key pair for signing
+            public_key, private_key = generate_key_pair()
+            signer = Ed25519Signer.from_private_key_b64(private_key)
+            
+            # Get contract to get content hash
+            contract_response = await client.get(f"/contracts/{contract_id}")
+            contract_data_response = contract_response.json()
+            content_hash = contract_data_response["content_hash"]
+            
+            # Create signature
+            signing_message = f"{contract_id}:{content_hash}:party1:provider"
+            signature = signer.sign(signing_message)
+            
+            # Sign contract
+            sign_data = {
+                "signer_id": "party1",
+                "signer_type": "provider",
+                "public_key": public_key,
+                "signature": signature
+            }
+            
+            response = await client.post(f"/contracts/{contract_id}/sign", json=sign_data)
+            assert response.status_code == 200
+            data = response.json()
+            assert len(data["signatures"]) == 1
+            assert data["signatures"][0]["signer_id"] == "party1"
+
+    @pytest.mark.asyncio
+    async def test_revoke_contract_endpoint(self, app):
+        """Test the revoke contract endpoint."""
+        import httpx
+        
+        # First create a contract
+        contract_data = {
+            "title": "Test Contract",
+            "description": "A test contract",
+            "clauses": [],
+            "parties": []
+        }
+        
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            # Create contract
+            create_response = await client.post("/contracts", json=contract_data)
+            contract_id = create_response.json()["id"]
+            
+            # Revoke contract
+            revoke_data = {
+                "reason": "Contract terms violated",
+                "revoked_by": "admin"
+            }
+            
+            response = await client.post(f"/contracts/{contract_id}/revoke", json=revoke_data)
+            assert response.status_code == 200
+            data = response.json()
+            assert data["state"] == "revoked"
+            assert data["revoked_at"] is not None
+
+    @pytest.mark.asyncio
+    async def test_contract_statistics_endpoint(self, app):
+        """Test the contract statistics endpoint."""
+        import httpx
+        
+        # Create some contracts
+        contract_data = {
+            "title": "Test Contract",
+            "description": "A test contract",
+            "clauses": [],
+            "parties": []
+        }
+        
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            # Create contracts
+            await client.post("/contracts", json=contract_data)
+            await client.post("/contracts", json=contract_data)
+            
+            # Get statistics
+            response = await client.get("/contracts/statistics")
+            assert response.status_code == 200
+            data = response.json()
+            assert "total_contracts" in data
+            assert "by_state" in data
+            assert "hipaa_compliant" in data
+            assert "signed_contracts" in data
+            assert data["total_contracts"] == 2