Phase 1 : Pluggable Policy Engine — design an interface that allows policies to be loaded/unloaded at runtime; each policy should define machine-checkable invariants and decision logic.

examples/policy_example.py

@@ -0,0 +1,248 @@
+"""
+FastMCP Policy Engine Example
+
+This example demonstrates the pluggable policy engine with:
+- Minimum Necessary Access Policy
+- RBAC Policy
+- Policy evaluation via HTTP endpoint
+- Hot reloading of policies
+"""
+
+import asyncio
+import json
+from pathlib import Path
+
+from fastmcp import FastMCP
+from fastmcp.policy import PolicyEngine
+from fastmcp.policy.policies import MinimumNecessaryAccessPolicy, RBACPolicy
+
+
+def create_policy_example():
+    """Create a FastMCP server with policy engine enabled."""
+    
+    # Create server
+    server = FastMCP("Policy Example Server")
+    
+    # Enable policy engine
+    policy_engine = server.enable_policy_engine()
+    
+    # Register built-in policies
+    policy_engine.register_policy(MinimumNecessaryAccessPolicy())
+    policy_engine.register_policy(RBACPolicy())
+    
+    # Add a simple tool that demonstrates policy evaluation
+    @server.tool
+    async def access_resource(user_id: str, action: str, resource_type: str) -> dict:
+        """Access a resource with policy evaluation."""
+        
+        # Create context for policy evaluation
+        context = {
+            "user": {
+                "id": user_id,
+                "roles": ["user"] if user_id != "admin" else ["admin"],
+                "permissions": ["read", "write"] if user_id != "admin" else ["*"]
+            },
+            "action": action,
+            "resource": {
+                "type": resource_type,
+                "id": f"{resource_type}_123",
+                "owner": user_id,
+                "visibility": "private"
+            }
+        }
+        
+        # Evaluate policies
+        decision = await policy_engine.evaluate(context)
+        
+        return {
+            "access_granted": decision.allow,
+            "reason": decision.reason,
+            "obligations": decision.obligations,
+            "proof": decision.proof
+        }
+    
+    return server
+
+
+async def demonstrate_policy_evaluation():
+    """Demonstrate policy evaluation with different scenarios."""
+    
+    server = create_policy_example()
+    policy_engine = server.get_policy_engine()
+    assert policy_engine is not None  # Ensure policy engine is enabled
+    
+    print("🔐 FastMCP Policy Engine Example")
+    print("=" * 50)
+    
+    # Test scenarios
+    scenarios = [
+        {
+            "name": "Regular user reading document",
+            "user_id": "user123",
+            "action": "read",
+            "resource_type": "document"
+        },
+        {
+            "name": "Regular user deleting sensitive data",
+            "user_id": "user123",
+            "action": "delete",
+            "resource_type": "user_data"
+        },
+        {
+            "name": "Admin accessing sensitive data",
+            "user_id": "admin",
+            "action": "delete",
+            "resource_type": "user_data"
+        },
+        {
+            "name": "User without roles",
+            "user_id": "guest",
+            "action": "read",
+            "resource_type": "document"
+        }
+    ]
+    
+    for scenario in scenarios:
+        print(f"\n📋 Scenario: {scenario['name']}")
+        print("-" * 40)
+        
+        # Create context
+        context = {
+            "user": {
+                "id": scenario["user_id"],
+                "roles": ["user"] if scenario["user_id"] != "admin" else ["admin"],
+                "permissions": ["read", "write"] if scenario["user_id"] != "admin" else ["*"]
+            },
+            "action": scenario["action"],
+            "resource": {
+                "type": scenario["resource_type"],
+                "id": f"{scenario['resource_type']}_123",
+                "owner": scenario["user_id"],
+                "visibility": "private"
+            }
+        }
+        
+        # Evaluate policies
+        decision = await policy_engine.evaluate(context)
+        
+        print(f"User: {scenario['user_id']}")
+        print(f"Action: {scenario['action']}")
+        print(f"Resource: {scenario['resource_type']}")
+        print(f"Decision: {'✅ ALLOW' if decision.allow else '❌ DENY'}")
+        print(f"Reason: {decision.reason}")
+        
+        if decision.obligations:
+            print("Obligations:")
+            for obligation in decision.obligations:
+                print(f"  - {obligation.get('type', 'unknown')}: {obligation.get('description', 'No description')}")
+        
+        if decision.proof:
+            print(f"Proof: {json.dumps(decision.proof, indent=2)}")
+
+
+def create_yaml_config():
+    """Create a YAML configuration file for policies."""
+    
+    config_content = """
+policies:
+  - name: custom_minimum_necessary
+    type: minimum_necessary
+    parameters:
+      required_justification: true
+      sensitive_actions:
+        - "delete"
+        - "admin"
+        - "privileged"
+      sensitive_resources:
+        - "user_data"
+        - "financial"
+        - "medical"
+  
+  - name: custom_rbac
+    type: rbac
+    parameters:
+      version: "1.0.0"
+      roles:
+        admin:
+          description: "Administrator with full access"
+          permissions: ["*"]
+        user:
+          description: "Regular user with basic access"
+          permissions: ["read", "write"]
+        guest:
+          description: "Guest user with read-only access"
+          permissions: ["read"]
+"""
+    
+    config_path = Path("policies.yaml")
+    config_path.write_text(config_content)
+    print(f"📄 Created YAML config: {config_path}")
+    return config_path
+
+
+async def demonstrate_hot_reload():
+    """Demonstrate hot reloading of policies."""
+    
+    print("\n🔄 Hot Reload Demonstration")
+    print("=" * 50)
+    
+    # Create server with policy engine
+    server = create_policy_example()
+    policy_engine = server.get_policy_engine()
+    assert policy_engine is not None  # Ensure policy engine is enabled
+    
+    # Register policy classes for YAML loading
+    policy_engine.registry.register_policy_class("minimum_necessary", MinimumNecessaryAccessPolicy)
+    policy_engine.registry.register_policy_class("rbac", RBACPolicy)
+    
+    # Create YAML config
+    config_path = create_yaml_config()
+    
+    # Load policies from YAML
+    print("Loading policies from YAML...")
+    policy_engine.registry.load_policies_from_yaml(config_path)
+    
+    # List loaded policies
+    policies = policy_engine.get_policy_metadata()
+    print(f"Loaded {len(policies)} policies:")
+    for policy in policies:
+        print(f"  - {policy['name']} ({policy['type']}) v{policy['version']}")
+    
+    # Demonstrate hot reload
+    print("\nHot reloading policies...")
+    policy_engine.registry.hot_reload_policies(config_path)
+    
+    # Clean up
+    config_path.unlink()
+    print("✅ Hot reload demonstration complete")
+
+
+def main():
+    """Run the policy engine example."""
+    
+    async def run_example():
+        # Demonstrate policy evaluation
+        await demonstrate_policy_evaluation()
+        
+        # Demonstrate hot reload
+        await demonstrate_hot_reload()
+        
+        print("\n🚀 Policy Engine Example Complete!")
+        print("\nTo test the HTTP endpoint:")
+        print("1. Run: fastmcp run examples/policy_example.py --transport http")
+        print("2. POST to /policy/evaluate with JSON body:")
+        print("""
+{
+  "context": {
+    "user": {"id": "user123", "roles": ["user"]},
+    "action": "read",
+    "resource": {"type": "document", "id": "doc123"}
+  }
+}
+        """)
+    
+    asyncio.run(run_example())
+
+
+if __name__ == "__main__":
+    main()

examples/smart_home/requirements.txt

@@ -0,0 +1,3 @@
+fastmcp@git+https://github.com/jlowin/fastmcp.git
+phue2
+

requirements-dev.txt

@@ -0,0 +1,33 @@
+# Core dependencies
+python-dotenv>=1.1.0
+exceptiongroup>=1.2.2
+httpx>=0.28.1
+mcp>=1.12.4,<2.0.0
+openapi-pydantic>=0.5.1
+rich>=13.9.4
+cyclopts>=3.0.0
+authlib>=1.5.2
+pydantic[email]>=2.11.7
+pyperclip>=1.9.0
+openapi-core>=0.19.5
+
+# Development dependencies
+copychat>=0.5.2
+dirty-equals>=0.9.0
+fastapi>=0.115.12
+ipython>=8.12.3
+pdbpp>=0.10.3
+pre-commit
+psutil
+pyinstrument>=5.0.2
+pyright>=1.1.389
+pytest>=8.3.3
+pytest-asyncio>=0.23.5
+pytest-cov>=6.1.1
+pytest-env>=1.1.5
+pytest-flakefinder
+pytest-httpx>=0.35.0
+pytest-report>=0.2.1
+pytest-timeout>=2.4.0
+pytest-xdist>=3.6.1
+ruff

requirements.txt

@@ -0,0 +1,11 @@
+python-dotenv>=1.1.0
+exceptiongroup>=1.2.2
+httpx>=0.28.1
+mcp>=1.12.4,<2.0.0
+openapi-pydantic>=0.5.1
+rich>=13.9.4
+cyclopts>=3.0.0
+authlib>=1.5.2
+pydantic[email]>=2.11.7
+pyperclip>=1.9.0
+openapi-core>=0.19.5

src/fastmcp/policy/__init__.py

@@ -0,0 +1,8 @@
+"""FastMCP Policy Engine - Pluggable authorization and access control."""
+
+from .decision import Decision
+from .engine import PolicyEngine
+from .policy import Policy
+from .registry import PolicyRegistry
+
+__all__ = ["Policy", "PolicyEngine", "PolicyRegistry", "Decision"]

src/fastmcp/policy/decision.py

@@ -0,0 +1,60 @@
+"""Policy decision result types."""
+
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class Decision:
+    """Result of a policy evaluation."""
+    
+    allow: bool
+    """Whether the action is allowed."""
+    
+    obligations: List[Dict[str, Any]]
+    """List of obligations that must be fulfilled."""
+    
+    reason: str
+    """Human-readable reason for the decision."""
+    
+    proof: Optional[Dict[str, Any]] = None
+    """Optional proof or evidence for the decision."""
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert decision to dictionary for JSON serialization."""
+        return {
+            "allow": self.allow,
+            "obligations": self.obligations,
+            "reason": self.reason,
+            "proof": self.proof,
+        }
+    
+    @classmethod
+    def allow_decision(
+        cls, 
+        reason: str = "Access granted", 
+        obligations: Optional[List[Dict[str, Any]]] = None,
+        proof: Optional[Dict[str, Any]] = None
+    ) -> "Decision":
+        """Create an allow decision."""
+        return cls(
+            allow=True,
+            obligations=obligations or [],
+            reason=reason,
+            proof=proof,
+        )
+    
+    @classmethod
+    def deny_decision(
+        cls, 
+        reason: str = "Access denied", 
+        obligations: Optional[List[Dict[str, Any]]] = None,
+        proof: Optional[Dict[str, Any]] = None
+    ) -> "Decision":
+        """Create a deny decision."""
+        return cls(
+            allow=False,
+            obligations=obligations or [],
+            reason=reason,
+            proof=proof,
+        )

src/fastmcp/policy/engine.py

@@ -0,0 +1,137 @@
+"""Policy engine for coordinating policy evaluation."""
+
+from typing import Any, Dict, List, Optional
+
+from fastmcp.utilities.logging import get_logger
+
+from .decision import Decision
+from .policy import Policy, PolicyContext
+from .registry import PolicyRegistry
+
+logger = get_logger(__name__)
+
+
+class PolicyEngine:
+    """Engine for evaluating policies."""
+    
+    def __init__(self, registry: Optional[PolicyRegistry] = None):
+        self.registry = registry or PolicyRegistry()
+        self._evaluation_order: List[str] = []
+    
+    def set_evaluation_order(self, policy_names: List[str]) -> None:
+        """Set the order in which policies should be evaluated.
+        
+        Args:
+            policy_names: List of policy names in evaluation order
+        """
+        self._evaluation_order = policy_names.copy()
+        logger.info(f"Set policy evaluation order: {policy_names}")
+    
+    async def evaluate(
+        self, 
+        context: Dict[str, Any], 
+        policy_names: Optional[List[str]] = None
+    ) -> Decision:
+        """Evaluate policies against the given context.
+        
+        Args:
+            context: The context for evaluation
+            policy_names: Optional list of policy names to evaluate.
+                         If None, evaluates all registered policies.
+        
+        Returns:
+            The final decision after evaluating all policies
+        """
+        if policy_names is None:
+            # Use evaluation order if set, otherwise use all policies
+            if self._evaluation_order:
+                policy_names = self._evaluation_order
+            else:
+                policy_names = list(self.registry._policies.keys())
+        
+        logger.debug(f"Evaluating policies: {policy_names}")
+        
+        # Evaluate each policy in order
+        for policy_name in policy_names:
+            policy = self.registry.get_policy(policy_name)
+            if not policy:
+                logger.warning(f"Policy not found: {policy_name}")
+                continue
+            
+            try:
+                decision = await policy.evaluate(context)
+                logger.debug(f"Policy {policy_name} decision: {decision.allow} - {decision.reason}")
+                
+                # If any policy denies, return deny decision
+                if not decision.allow:
+                    return decision
+                    
+            except Exception as e:
+                logger.error(f"Error evaluating policy {policy_name}: {e}")
+                return Decision.deny_decision(
+                    reason=f"Policy evaluation error: {e}",
+                    proof={"policy": policy_name, "error": str(e)}
+                )
+        
+        # All policies allowed
+        return Decision.allow_decision(
+            reason="All policies evaluated successfully",
+            proof={"evaluated_policies": policy_names}
+        )
+    
+    async def evaluate_single_policy(
+        self, 
+        policy_name: str, 
+        context: Dict[str, Any]
+    ) -> Optional[Decision]:
+        """Evaluate a single policy.
+        
+        Args:
+            policy_name: The name of the policy to evaluate
+            context: The context for evaluation
+            
+        Returns:
+            The decision, or None if policy not found
+        """
+        policy = self.registry.get_policy(policy_name)
+        if not policy:
+            logger.warning(f"Policy not found: {policy_name}")
+            return None
+        
+        try:
+            decision = await policy.evaluate(context)
+            logger.debug(f"Single policy {policy_name} decision: {decision.allow} - {decision.reason}")
+            return decision
+        except Exception as e:
+            logger.error(f"Error evaluating policy {policy_name}: {e}")
+            return Decision.deny_decision(
+                reason=f"Policy evaluation error: {e}",
+                proof={"policy": policy_name, "error": str(e)}
+            )
+    
+    def get_policy_metadata(self) -> List[Dict[str, Any]]:
+        """Get metadata for all registered policies.
+        
+        Returns:
+            List of policy metadata dictionaries
+        """
+        return self.registry.list_policies()
+    
+    def register_policy(self, policy: Policy) -> None:
+        """Register a policy with the engine.
+        
+        Args:
+            policy: The policy to register
+        """
+        self.registry.register_policy(policy)
+    
+    def unregister_policy(self, name: str) -> Optional[Policy]:
+        """Unregister a policy from the engine.
+        
+        Args:
+            name: The name of the policy to unregister
+            
+        Returns:
+            The unregistered policy, or None if not found
+        """
+        return self.registry.unregister_policy(name)

src/fastmcp/policy/policies/__init__.py

@@ -0,0 +1,6 @@
+"""Built-in policy implementations."""
+
+from .minimum_necessary import MinimumNecessaryAccessPolicy
+from .rbac import RBACPolicy
+
+__all__ = ["MinimumNecessaryAccessPolicy", "RBACPolicy"]

src/fastmcp/policy/policies/minimum_necessary.py

@@ -0,0 +1,144 @@
+"""Minimum Necessary Access Policy implementation."""
+
+from typing import Any, Dict, List, Set, Optional
+
+from ..policy import Policy
+from ..decision import Decision
+
+
+class MinimumNecessaryAccessPolicy(Policy):
+    """Policy that enforces minimum necessary access principles."""
+    
+    def __init__(
+        self, 
+        name: str = "minimum_necessary_access",
+        version: str = "1.0.0",
+        sensitive_actions: Optional[List[str]] = None,
+        sensitive_resources: Optional[List[str]] = None,
+        required_justification: bool = True
+    ):
+        super().__init__(name, version)
+        self.sensitive_actions = set(sensitive_actions or [
+            "delete", "modify", "admin", "root", "sudo", "privileged"
+        ])
+        self.sensitive_resources = set(sensitive_resources or [
+            "user_data", "financial", "medical", "personal", "confidential"
+        ])
+        self.required_justification = required_justification
+    
+    async def evaluate(self, context: Dict[str, Any]) -> Decision:
+        """Evaluate minimum necessary access policy.
+        
+        Args:
+            context: The context containing user, action, resource information
+            
+        Returns:
+            Decision indicating whether access is allowed
+        """
+        user = context.get("user", {})
+        action = context.get("action", "")
+        resource = context.get("resource", {})
+        
+        # Check if action is sensitive
+        action_lower = action.lower()
+        is_sensitive_action = any(
+            sensitive in action_lower for sensitive in self.sensitive_actions
+        )
+        
+        # Check if resource is sensitive
+        resource_type = resource.get("type", "")
+        resource_tags = resource.get("tags", [])
+        is_sensitive_resource = (
+            resource_type in self.sensitive_resources or
+            any(tag in self.sensitive_resources for tag in resource_tags)
+        )
+        
+        # If neither action nor resource is sensitive, allow
+        if not is_sensitive_action and not is_sensitive_resource:
+            return Decision.allow_decision(
+                reason="Action and resource are not sensitive",
+                proof={
+                    "action_sensitive": False,
+                    "resource_sensitive": False
+                }
+            )
+        
+        # Check for justification if required
+        if self.required_justification:
+            justification = context.get("justification", "")
+            if not justification or len(justification.strip()) < 10:
+                return Decision.deny_decision(
+                    reason="Sensitive operation requires justification",
+                    obligations=[
+                        {
+                            "type": "provide_justification",
+                            "description": "Provide a detailed justification for this sensitive operation"
+                        }
+                    ],
+                    proof={
+                        "action_sensitive": is_sensitive_action,
+                        "resource_sensitive": is_sensitive_resource,
+                        "justification_provided": bool(justification),
+                        "justification_length": len(justification) if justification else 0
+                    }
+                )
+        
+        # Check user permissions
+        user_roles = user.get("roles", [])
+        user_permissions = user.get("permissions", [])
+        
+        # Allow if user has explicit permission
+        if "admin" in user_roles or "privileged" in user_permissions:
+            return Decision.allow_decision(
+                reason="User has privileged access",
+                obligations=[
+                    {
+                        "type": "audit_log",
+                        "description": "Log this sensitive operation for audit purposes"
+                    }
+                ],
+                proof={
+                    "user_roles": user_roles,
+                    "user_permissions": user_permissions,
+                    "action_sensitive": is_sensitive_action,
+                    "resource_sensitive": is_sensitive_resource
+                }
+            )
+        
+        # Check for time-based restrictions
+        time_context = context.get("time", {})
+        current_hour = time_context.get("hour", 0)
+        
+        # Restrict sensitive operations during off-hours (example: 10 PM to 6 AM)
+        if is_sensitive_action and (current_hour >= 22 or current_hour < 6):
+            return Decision.deny_decision(
+                reason="Sensitive operations restricted during off-hours",
+                obligations=[
+                    {
+                        "type": "schedule_operation",
+                        "description": "Schedule this operation during business hours"
+                    }
+                ],
+                proof={
+                    "current_hour": current_hour,
+                    "off_hours": True,
+                    "action_sensitive": True
+                }
+            )
+        
+        # Default deny for sensitive operations without proper authorization
+        return Decision.deny_decision(
+            reason="Insufficient permissions for sensitive operation",
+            obligations=[
+                {
+                    "type": "request_approval",
+                    "description": "Request approval from administrator"
+                }
+            ],
+            proof={
+                "action_sensitive": is_sensitive_action,
+                "resource_sensitive": is_sensitive_resource,
+                "user_roles": user_roles,
+                "user_permissions": user_permissions
+            }
+        )

src/fastmcp/policy/policies/rbac.py

@@ -0,0 +1,233 @@
+"""Role-Based Access Control (RBAC) Policy implementation."""
+
+from typing import Any, Dict, List, Optional, Set
+
+from ..policy import Policy
+from ..decision import Decision
+
+
+class RBACPolicy(Policy):
+    """Policy that implements Role-Based Access Control."""
+    
+    def __init__(
+        self,
+        name: str = "rbac",
+        version: str = "1.0.0",
+        roles: Optional[Dict[str, Dict[str, Any]]] = None,
+        permissions: Optional[Dict[str, List[str]]] = None,
+        role_hierarchy: Optional[Dict[str, List[str]]] = None
+    ):
+        super().__init__(name, version)
+        
+        # Default roles and permissions
+        self.roles = roles or {
+            "admin": {
+                "description": "Administrator with full access",
+                "permissions": ["*"]
+            },
+            "user": {
+                "description": "Regular user with basic access",
+                "permissions": ["read", "write"]
+            },
+            "guest": {
+                "description": "Guest user with read-only access",
+                "permissions": ["read"]
+            }
+        }
+        
+        # Custom permissions mapping
+        self.permissions = permissions or {
+            "read": ["get", "list", "view", "read"],
+            "write": ["create", "update", "modify", "write"],
+            "delete": ["remove", "delete", "destroy"],
+            "admin": ["admin", "manage", "configure", "privileged"]
+        }
+        
+        # Role hierarchy (inheritance)
+        self.role_hierarchy = role_hierarchy or {
+            "admin": ["user", "guest"],
+            "user": ["guest"]
+        }
+    
+    def _get_user_permissions(self, user_roles: List[str]) -> Set[str]:
+        """Get all permissions for a user based on their roles and hierarchy.
+        
+        Args:
+            user_roles: List of user roles
+            
+        Returns:
+            Set of all permissions the user has
+        """
+        all_permissions = set()
+        
+        for role in user_roles:
+            if role not in self.roles:
+                continue
+            
+            # Get direct permissions for this role
+            role_permissions = self.roles[role].get("permissions", [])
+            if "*" in role_permissions:
+                # Wildcard permission - user has all permissions
+                return {"*"}
+            
+            all_permissions.update(role_permissions)
+            
+            # Get inherited permissions from role hierarchy
+            inherited_roles = self.role_hierarchy.get(role, [])
+            for inherited_role in inherited_roles:
+                if inherited_role in self.roles:
+                    inherited_permissions = self.roles[inherited_role].get("permissions", [])
+                    all_permissions.update(inherited_permissions)
+        
+        return all_permissions
+    
+    def _check_permission(self, user_permissions: Set[str], required_action: str) -> bool:
+        """Check if user has permission for the required action.
+        
+        Args:
+            user_permissions: Set of user permissions
+            required_action: The action being performed
+            
+        Returns:
+            True if user has permission, False otherwise
+        """
+        # Check for wildcard permission
+        if "*" in user_permissions:
+            return True
+        
+        # Check direct permission match
+        if required_action in user_permissions:
+            return True
+        
+        # Check permission mappings
+        for permission, actions in self.permissions.items():
+            if permission in user_permissions and required_action in actions:
+                return True
+        
+        return False
+    
+    async def evaluate(self, context: Dict[str, Any]) -> Decision:
+        """Evaluate RBAC policy.
+        
+        Args:
+            context: The context containing user, action, resource information
+            
+        Returns:
+            Decision indicating whether access is allowed
+        """
+        user = context.get("user", {})
+        action = context.get("action", "")
+        resource = context.get("resource", {})
+        
+        # Get user roles
+        user_roles = user.get("roles", [])
+        if not user_roles:
+            return Decision.deny_decision(
+                reason="User has no assigned roles",
+                proof={
+                    "user_roles": user_roles,
+                    "action": action
+                }
+            )
+        
+        # Get all user permissions
+        user_permissions = self._get_user_permissions(user_roles)
+        
+        # Check if user has permission for the action
+        has_permission = self._check_permission(user_permissions, action)
+        
+        if has_permission:
+            # Check resource-specific restrictions
+            resource_type = resource.get("type", "")
+            resource_owner = resource.get("owner", "")
+            user_id = user.get("id", "")
+            
+            # Allow if user owns the resource or has admin role
+            if (resource_owner == user_id or 
+                "admin" in user_roles or 
+                "*" in user_permissions):
+                return Decision.allow_decision(
+                    reason="User has permission and owns resource or is admin",
+                    obligations=[
+                        {
+                            "type": "audit_log",
+                            "description": "Log this RBAC-authorized operation"
+                        }
+                    ],
+                    proof={
+                        "user_roles": user_roles,
+                        "user_permissions": list(user_permissions),
+                        "action": action,
+                        "resource_owner": resource_owner,
+                        "user_id": user_id,
+                        "permission_check": True
+                    }
+                )
+            
+            # Check if resource is public/shared
+            resource_visibility = resource.get("visibility", "private")
+            if resource_visibility in ["public", "shared"]:
+                return Decision.allow_decision(
+                    reason="User has permission and resource is public/shared",
+                    proof={
+                        "user_roles": user_roles,
+                        "user_permissions": list(user_permissions),
+                        "action": action,
+                        "resource_visibility": resource_visibility,
+                        "permission_check": True
+                    }
+                )
+            
+            # Check for explicit resource permissions
+            resource_permissions = resource.get("permissions", {})
+            if user_id in resource_permissions:
+                user_resource_permissions = resource_permissions[user_id]
+                if action in user_resource_permissions or "*" in user_resource_permissions:
+                    return Decision.allow_decision(
+                        reason="User has explicit resource permission",
+                        proof={
+                            "user_roles": user_roles,
+                            "user_permissions": list(user_permissions),
+                            "action": action,
+                            "resource_permissions": user_resource_permissions,
+                            "permission_check": True
+                        }
+                    )
+            
+            # Deny access to private resource without ownership
+            return Decision.deny_decision(
+                reason="User lacks permission for this private resource",
+                obligations=[
+                    {
+                        "type": "request_access",
+                        "description": "Request access from resource owner"
+                    }
+                ],
+                proof={
+                    "user_roles": user_roles,
+                    "user_permissions": list(user_permissions),
+                    "action": action,
+                    "resource_owner": resource_owner,
+                    "user_id": user_id,
+                    "resource_visibility": resource_visibility,
+                    "permission_check": True,
+                    "ownership_check": False
+                }
+            )
+        else:
+            # User doesn't have permission for the action
+            return Decision.deny_decision(
+                reason="User lacks permission for this action",
+                obligations=[
+                    {
+                        "type": "request_permission",
+                        "description": "Request permission from administrator"
+                    }
+                ],
+                proof={
+                    "user_roles": user_roles,
+                    "user_permissions": list(user_permissions),
+                    "action": action,
+                    "permission_check": False
+                }
+            )

src/fastmcp/policy/policy.py

@@ -0,0 +1,63 @@
+"""Policy interface and base classes."""
+
+from abc import ABC, abstractmethod
+from typing import Any, Dict, Optional
+
+from .decision import Decision
+
+
+class Policy(ABC):
+    """Base class for all policies."""
+    
+    def __init__(self, name: str, version: str = "1.0.0"):
+        self.name = name
+        self.version = version
+    
+    @abstractmethod
+    async def evaluate(self, context: Dict[str, Any]) -> Decision:
+        """Evaluate the policy against the given context.
+        
+        Args:
+            context: The context containing information about the request,
+                    user, resource, action, etc.
+        
+        Returns:
+            A Decision object indicating whether access is allowed or denied.
+        """
+        pass
+    
+    def get_metadata(self) -> Dict[str, Any]:
+        """Get policy metadata."""
+        return {
+            "name": self.name,
+            "version": self.version,
+            "type": self.__class__.__name__,
+        }
+
+
+class PolicyContext:
+    """Context for policy evaluation."""
+    
+    def __init__(
+        self,
+        user: Optional[Dict[str, Any]] = None,
+        resource: Optional[Dict[str, Any]] = None,
+        action: Optional[str] = None,
+        environment: Optional[Dict[str, Any]] = None,
+        **kwargs
+    ):
+        self.user = user or {}
+        self.resource = resource or {}
+        self.action = action
+        self.environment = environment or {}
+        self.extra = kwargs
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert context to dictionary."""
+        return {
+            "user": self.user,
+            "resource": self.resource,
+            "action": self.action,
+            "environment": self.environment,
+            **self.extra,
+        }

src/fastmcp/policy/registry.py

@@ -0,0 +1,189 @@
+"""Policy registry for managing policies at runtime."""
+
+import importlib
+import importlib.metadata
+import json
+import yaml
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Type
+
+from fastmcp.utilities.logging import get_logger
+
+from .policy import Policy
+
+logger = get_logger(__name__)
+
+
+class PolicyRegistry:
+    """Registry for managing policies at runtime."""
+    
+    def __init__(self):
+        self._policies: Dict[str, Policy] = {}
+        self._policy_classes: Dict[str, Type[Policy]] = {}
+    
+    def register_policy(self, policy: Policy) -> None:
+        """Register a policy instance.
+        
+        Args:
+            policy: The policy instance to register
+        """
+        self._policies[policy.name] = policy
+        logger.info(f"Registered policy: {policy.name} v{policy.version}")
+    
+    def unregister_policy(self, name: str) -> Optional[Policy]:
+        """Unregister a policy by name.
+        
+        Args:
+            name: The name of the policy to unregister
+            
+        Returns:
+            The unregistered policy, or None if not found
+        """
+        policy = self._policies.pop(name, None)
+        if policy:
+            logger.info(f"Unregistered policy: {name}")
+        return policy
+    
+    def get_policy(self, name: str) -> Optional[Policy]:
+        """Get a policy by name.
+        
+        Args:
+            name: The name of the policy
+            
+        Returns:
+            The policy instance, or None if not found
+        """
+        return self._policies.get(name)
+    
+    def list_policies(self) -> List[Dict[str, Any]]:
+        """List all registered policies.
+        
+        Returns:
+            List of policy metadata dictionaries
+        """
+        return [policy.get_metadata() for policy in self._policies.values()]
+    
+    def load_policy_from_entry_point(self, entry_point_name: str) -> None:
+        """Load policies from entry points.
+        
+        Args:
+            entry_point_name: The entry point name to load from
+        """
+        try:
+            entry_points = importlib.metadata.entry_points()
+            if hasattr(entry_points, 'select'):
+                # Python 3.10+
+                policy_entry_points = entry_points.select(group=entry_point_name)
+            else:
+                # Python 3.8-3.9
+                policy_entry_points = entry_points.get(entry_point_name, [])
+            
+            for entry_point in policy_entry_points:
+                try:
+                    policy_class = entry_point.load()
+                    if issubclass(policy_class, Policy):
+                        # Create policy with default name from entry point
+                        policy = policy_class(name=entry_point.name)
+                        self.register_policy(policy)
+                    else:
+                        logger.warning(f"Entry point {entry_point.name} does not return a Policy class")
+                except Exception as e:
+                    logger.error(f"Failed to load policy from entry point {entry_point.name}: {e}")
+        except Exception as e:
+            logger.error(f"Failed to load policies from entry points: {e}")
+    
+    def load_policies_from_yaml(self, yaml_path: Path) -> None:
+        """Load policies from a YAML specification file.
+        
+        Args:
+            yaml_path: Path to the YAML file
+        """
+        try:
+            with open(yaml_path, 'r') as f:
+                config = yaml.safe_load(f)
+            
+            policies_config = config.get('policies', [])
+            for policy_config in policies_config:
+                try:
+                    policy_name = policy_config['name']
+                    policy_type = policy_config['type']
+                    policy_params = policy_config.get('parameters', {})
+                    
+                    # Get the policy class
+                    policy_class = self._policy_classes.get(policy_type)
+                    if not policy_class:
+                        logger.error(f"Unknown policy type: {policy_type}")
+                        continue
+                    
+                    # Create and register the policy
+                    policy = policy_class(name=policy_name, **policy_params)
+                    self.register_policy(policy)
+                    
+                except KeyError as e:
+                    logger.error(f"Missing required field in policy config: {e}")
+                except Exception as e:
+                    logger.error(f"Failed to load policy from config: {e}")
+                    
+        except Exception as e:
+            logger.error(f"Failed to load policies from YAML file {yaml_path}: {e}")
+    
+    def register_policy_class(self, name: str, policy_class: Type[Policy]) -> None:
+        """Register a policy class for dynamic instantiation.
+        
+        Args:
+            name: The name to register the policy class under
+            policy_class: The policy class to register
+        """
+        if not issubclass(policy_class, Policy):
+            raise ValueError(f"Class {policy_class} must inherit from Policy")
+        
+        self._policy_classes[name] = policy_class
+        logger.info(f"Registered policy class: {name}")
+    
+    def create_policy_from_config(self, config: Dict[str, Any]) -> Optional[Policy]:
+        """Create a policy instance from configuration.
+        
+        Args:
+            config: Configuration dictionary
+            
+        Returns:
+            The created policy instance, or None if creation failed
+        """
+        try:
+            policy_type = config.get('type')
+            if not policy_type:
+                logger.error("Policy config missing 'type' field")
+                return None
+            
+            policy_class = self._policy_classes.get(policy_type)
+            if not policy_class:
+                logger.error(f"Unknown policy type: {policy_type}")
+                return None
+            
+            policy_params = config.get('parameters', {})
+            policy = policy_class(**policy_params)
+            return policy
+            
+        except Exception as e:
+            logger.error(f"Failed to create policy from config: {e}")
+            return None
+    
+    def hot_reload_policies(self, yaml_path: Path) -> None:
+        """Hot reload policies from YAML file.
+        
+        Args:
+            yaml_path: Path to the YAML file
+        """
+        logger.info(f"Hot reloading policies from {yaml_path}")
+        
+        # Clear existing policies
+        self._policies.clear()
+        
+        # Reload from entry points
+        self.load_policy_from_entry_point("fastmcp.policies")
+        
+        # Reload from YAML
+        if yaml_path.exists():
+            self.load_policies_from_yaml(yaml_path)
+        
+        logger.info(f"Hot reload complete. {len(self._policies)} policies loaded")

src/fastmcp/server/policy_routes.py

@@ -0,0 +1,87 @@
+"""Policy evaluation HTTP routes."""
+
+from typing import Any, Dict
+
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.routing import Route
+
+from fastmcp.policy import PolicyEngine
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+async def policy_evaluate_endpoint(request: Request) -> JSONResponse:
+    """HTTP endpoint for policy evaluation.
+    
+    Expected JSON body:
+    {
+        "context": {
+            "user": {...},
+            "action": "read",
+            "resource": {...},
+            ...
+        },
+        "policy_names": ["policy1", "policy2"]  // optional
+    }
+    """
+    try:
+        # Parse request body
+        body = await request.json()
+        
+        # Get policy engine from request state
+        policy_engine: PolicyEngine = request.app.state.policy_engine
+        
+        # Extract context and optional policy names
+        context = body.get("context", {})
+        policy_names = body.get("policy_names")
+        
+        if not context:
+            return JSONResponse(
+                status_code=400,
+                content={
+                    "error": "Missing required 'context' field",
+                    "reason": "Policy evaluation requires a context object"
+                }
+            )
+        
+        # Evaluate policies
+        decision = await policy_engine.evaluate(context, policy_names)
+        
+        # Return structured decision
+        return JSONResponse(
+            status_code=200,
+            content=decision.to_dict()
+        )
+        
+    except Exception as e:
+        logger.error(f"Policy evaluation error: {e}")
+        return JSONResponse(
+            status_code=500,
+            content={
+                "error": "Policy evaluation failed",
+                "reason": str(e)
+            }
+        )
+
+
+def create_policy_evaluate_route(policy_engine: PolicyEngine) -> Route:
+    """Create the policy evaluation route.
+    
+    Args:
+        policy_engine: The policy engine instance
+        
+    Returns:
+        Starlette Route for policy evaluation
+    """
+    async def endpoint_with_engine(request: Request) -> JSONResponse:
+        # Store policy engine in app state for access in endpoint
+        request.app.state.policy_engine = policy_engine
+        return await policy_evaluate_endpoint(request)
+    
+    return Route(
+        path="/policy/evaluate",
+        endpoint=endpoint_with_engine,
+        methods=["POST"]
+    )

src/fastmcp/server/server.py

@@ -14,7 +14,7 @@ from contextlib import (
 from dataclasses import dataclass
 from functools import partial
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Generic, Literal, cast, overload
+from typing import TYPE_CHECKING, Any, Generic, Literal, Optional, cast, overload
 
 import anyio
 import httpx
@@ -62,6 +62,7 @@ from fastmcp.settings import Settings
 from fastmcp.tools import ToolManager
 from fastmcp.tools.tool import FunctionTool, Tool, ToolResult
 from fastmcp.tools.tool_transform import ToolTransformConfig
+from fastmcp.policy import PolicyEngine
 from fastmcp.utilities.cli import log_server_banner
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.logging import get_logger
@@ -173,6 +174,7 @@ class FastMCP(Generic[LifespanResultT]):
 
         self._additional_http_routes: list[BaseRoute] = []
         self._mounted_servers: list[MountedServer] = []
+        self._policy_engine: Optional[PolicyEngine] = None
         self._tool_manager = ToolManager(
             duplicate_behavior=on_duplicate_tools,
             mask_error_details=mask_error_details,
@@ -487,12 +489,43 @@ class FastMCP(Generic[LifespanResultT]):
         """
         routes = list(self._additional_http_routes)
 
+        # Add policy evaluation endpoint if policy engine is configured
+        if self._policy_engine is not None:
+            from fastmcp.server.policy_routes import create_policy_evaluate_route
+            
+            policy_route = create_policy_evaluate_route(self._policy_engine)
+            routes.append(policy_route)
+
         # Recursively get routes from mounted servers
         for mounted_server in self._mounted_servers:
             mounted_routes = mounted_server.server._get_additional_http_routes()
             routes.extend(mounted_routes)
 
         return routes
+    
+    def enable_policy_engine(self, policy_engine: Optional[PolicyEngine] = None) -> PolicyEngine:
+        """Enable the policy engine for this server.
+        
+        Args:
+            policy_engine: Optional policy engine instance. If None, creates a new one.
+            
+        Returns:
+            The policy engine instance
+        """
+        if policy_engine is None:
+            policy_engine = PolicyEngine()
+        
+        self._policy_engine = policy_engine
+        logger.info("Policy engine enabled for server")
+        return policy_engine
+    
+    def get_policy_engine(self) -> Optional[PolicyEngine]:
+        """Get the policy engine instance.
+        
+        Returns:
+            The policy engine instance, or None if not enabled
+        """
+        return self._policy_engine
 
     async def _mcp_list_tools(self) -> list[MCPTool]:
         logger.debug("Handler called: list_tools")

tests/policy/__init__.py

@@ -0,0 +1 @@
+"""Policy engine tests."""

tests/policy/test_policy_engine.py

@@ -0,0 +1,333 @@
+"""Tests for the policy engine."""
+
+import pytest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from fastmcp.policy import PolicyEngine, PolicyRegistry, Decision
+from fastmcp.policy.policies import MinimumNecessaryAccessPolicy, RBACPolicy
+
+
+class TestPolicyEngine:
+    """Test the policy engine functionality."""
+    
+    @pytest.fixture
+    def policy_engine(self):
+        """Create a policy engine for testing."""
+        return PolicyEngine()
+    
+    @pytest.fixture
+    def sample_context(self):
+        """Create a sample context for testing."""
+        return {
+            "user": {
+                "id": "user123",
+                "roles": ["user"],
+                "permissions": ["read", "write"]
+            },
+            "action": "read",
+            "resource": {
+                "type": "document",
+                "id": "doc123",
+                "owner": "user123",
+                "visibility": "private"
+            }
+        }
+    
+    def test_policy_engine_initialization(self, policy_engine):
+        """Test policy engine initialization."""
+        assert policy_engine.registry is not None
+        assert isinstance(policy_engine.registry, PolicyRegistry)
+        assert policy_engine._evaluation_order == []
+    
+    def test_set_evaluation_order(self, policy_engine):
+        """Test setting evaluation order."""
+        order = ["policy1", "policy2", "policy3"]
+        policy_engine.set_evaluation_order(order)
+        assert policy_engine._evaluation_order == order
+    
+    @pytest.mark.asyncio
+    async def test_evaluate_no_policies(self, policy_engine, sample_context):
+        """Test evaluation when no policies are registered."""
+        decision = await policy_engine.evaluate(sample_context)
+        assert decision.allow is True
+        assert "All policies evaluated successfully" in decision.reason
+    
+    @pytest.mark.asyncio
+    async def test_evaluate_single_policy(self, policy_engine, sample_context):
+        """Test evaluation of a single policy."""
+        # Register a policy
+        policy = MinimumNecessaryAccessPolicy()
+        policy_engine.register_policy(policy)
+        
+        # Evaluate
+        decision = await policy_engine.evaluate(sample_context)
+        assert decision.allow is True
+        # The policy engine returns its own response, not the individual policy response
+        assert "All policies evaluated successfully" in decision.reason
+    
+    @pytest.mark.asyncio
+    async def test_evaluate_specific_policies(self, policy_engine, sample_context):
+        """Test evaluation of specific policies."""
+        # Register multiple policies
+        policy1 = MinimumNecessaryAccessPolicy(name="policy1")
+        policy2 = RBACPolicy(name="policy2")
+        policy_engine.register_policy(policy1)
+        policy_engine.register_policy(policy2)
+        
+        # Evaluate only policy1
+        decision = await policy_engine.evaluate(sample_context, ["policy1"])
+        assert decision.allow is True
+    
+    @pytest.mark.asyncio
+    async def test_evaluate_policy_denial(self, policy_engine):
+        """Test evaluation when a policy denies access."""
+        # Create context with sensitive action
+        context = {
+            "user": {"roles": ["user"]},
+            "action": "delete",
+            "resource": {"type": "user_data"}
+        }
+        
+        # Register minimum necessary policy
+        policy = MinimumNecessaryAccessPolicy()
+        policy_engine.register_policy(policy)
+        
+        # Evaluate
+        decision = await policy_engine.evaluate(context)
+        assert decision.allow is False
+        assert "requires justification" in decision.reason
+    
+    @pytest.mark.asyncio
+    async def test_evaluate_single_policy_method(self, policy_engine, sample_context):
+        """Test evaluate_single_policy method."""
+        # Register a policy
+        policy = MinimumNecessaryAccessPolicy(name="test_policy")
+        policy_engine.register_policy(policy)
+        
+        # Evaluate single policy
+        decision = await policy_engine.evaluate_single_policy("test_policy", sample_context)
+        assert decision is not None
+        assert decision.allow is True
+    
+    @pytest.mark.asyncio
+    async def test_evaluate_single_policy_not_found(self, policy_engine, sample_context):
+        """Test evaluate_single_policy with non-existent policy."""
+        decision = await policy_engine.evaluate_single_policy("non_existent", sample_context)
+        assert decision is None
+    
+    def test_get_policy_metadata(self, policy_engine):
+        """Test getting policy metadata."""
+        # Register policies
+        policy1 = MinimumNecessaryAccessPolicy(name="policy1")
+        policy2 = RBACPolicy(name="policy2")
+        policy_engine.register_policy(policy1)
+        policy_engine.register_policy(policy2)
+        
+        metadata = policy_engine.get_policy_metadata()
+        assert len(metadata) == 2
+        assert any(p["name"] == "policy1" for p in metadata)
+        assert any(p["name"] == "policy2" for p in metadata)
+    
+    def test_register_unregister_policy(self, policy_engine):
+        """Test policy registration and unregistration."""
+        policy = MinimumNecessaryAccessPolicy(name="test_policy")
+        
+        # Register
+        policy_engine.register_policy(policy)
+        assert policy_engine.registry.get_policy("test_policy") is not None
+        
+        # Unregister
+        unregistered = policy_engine.unregister_policy("test_policy")
+        assert unregistered is not None
+        assert unregistered.name == "test_policy"
+        assert policy_engine.registry.get_policy("test_policy") is None
+
+
+class TestPolicyRegistry:
+    """Test the policy registry functionality."""
+    
+    @pytest.fixture
+    def registry(self):
+        """Create a policy registry for testing."""
+        return PolicyRegistry()
+    
+    def test_registry_initialization(self, registry):
+        """Test registry initialization."""
+        assert registry._policies == {}
+        assert registry._policy_classes == {}
+    
+    def test_register_policy(self, registry):
+        """Test policy registration."""
+        policy = MinimumNecessaryAccessPolicy()
+        registry.register_policy(policy)
+        assert "minimum_necessary_access" in registry._policies
+    
+    def test_unregister_policy(self, registry):
+        """Test policy unregistration."""
+        policy = MinimumNecessaryAccessPolicy()
+        registry.register_policy(policy)
+        
+        unregistered = registry.unregister_policy("minimum_necessary_access")
+        assert unregistered is not None
+        assert "minimum_necessary_access" not in registry._policies
+    
+    def test_get_policy(self, registry):
+        """Test getting a policy."""
+        policy = MinimumNecessaryAccessPolicy()
+        registry.register_policy(policy)
+        
+        retrieved = registry.get_policy("minimum_necessary_access")
+        assert retrieved is not None
+        assert retrieved.name == "minimum_necessary_access"
+    
+    def test_list_policies(self, registry):
+        """Test listing policies."""
+        policy1 = MinimumNecessaryAccessPolicy(name="policy1")
+        policy2 = RBACPolicy(name="policy2")
+        registry.register_policy(policy1)
+        registry.register_policy(policy2)
+        
+        policies = registry.list_policies()
+        assert len(policies) == 2
+        assert any(p["name"] == "policy1" for p in policies)
+        assert any(p["name"] == "policy2" for p in policies)
+    
+    def test_register_policy_class(self, registry):
+        """Test registering a policy class."""
+        registry.register_policy_class("test_policy", MinimumNecessaryAccessPolicy)
+        assert "test_policy" in registry._policy_classes
+        assert registry._policy_classes["test_policy"] == MinimumNecessaryAccessPolicy
+    
+    def test_register_invalid_policy_class(self, registry):
+        """Test registering an invalid policy class."""
+        with pytest.raises(ValueError):
+            registry.register_policy_class("invalid", str)
+    
+    def test_create_policy_from_config(self, registry):
+        """Test creating policy from configuration."""
+        registry.register_policy_class("test_policy", MinimumNecessaryAccessPolicy)
+        
+        config = {
+            "type": "test_policy",
+            "parameters": {
+                "name": "config_policy",
+                "required_justification": False
+            }
+        }
+        
+        policy = registry.create_policy_from_config(config)
+        assert policy is not None
+        assert policy.name == "config_policy"
+        assert isinstance(policy, MinimumNecessaryAccessPolicy)
+    
+    def test_create_policy_from_invalid_config(self, registry):
+        """Test creating policy from invalid configuration."""
+        config = {"type": "non_existent"}
+        policy = registry.create_policy_from_config(config)
+        assert policy is None
+
+
+class TestPolicyLoadAndReload:
+    """Test policy loading and hot-reload functionality."""
+    
+    @pytest.fixture
+    def registry(self):
+        """Create a policy registry for testing."""
+        return PolicyRegistry()
+    
+    @pytest.fixture
+    def yaml_config_file(self, tmp_path):
+        """Create a temporary YAML config file."""
+        config_content = """
+policies:
+  - name: yaml_policy1
+    type: minimum_necessary
+    parameters:
+      required_justification: false
+  - name: yaml_policy2
+    type: rbac
+    parameters:
+      version: "1.0.0"
+"""
+        config_file = tmp_path / "policies.yaml"
+        config_file.write_text(config_content)
+        return config_file
+    
+    def test_load_policies_from_yaml(self, registry, yaml_config_file):
+        """Test loading policies from YAML file."""
+        # Register policy classes
+        registry.register_policy_class("minimum_necessary", MinimumNecessaryAccessPolicy)
+        registry.register_policy_class("rbac", RBACPolicy)
+        
+        # Load from YAML
+        registry.load_policies_from_yaml(yaml_config_file)
+        
+        # Check that policies were loaded
+        assert registry.get_policy("yaml_policy1") is not None
+        assert registry.get_policy("yaml_policy2") is not None
+    
+    def test_hot_reload_policies(self, registry, yaml_config_file):
+        """Test hot reloading policies."""
+        # Register policy classes
+        registry.register_policy_class("minimum_necessary", MinimumNecessaryAccessPolicy)
+        registry.register_policy_class("rbac", RBACPolicy)
+        
+        # Initial load
+        registry.load_policies_from_yaml(yaml_config_file)
+        initial_count = len(registry._policies)
+        
+        # Hot reload
+        registry.hot_reload_policies(yaml_config_file)
+        
+        # Check that policies were reloaded
+        assert len(registry._policies) == initial_count
+        assert registry.get_policy("yaml_policy1") is not None
+        assert registry.get_policy("yaml_policy2") is not None
+    
+    @patch('importlib.metadata.entry_points')
+    def test_load_policy_from_entry_point(self, mock_entry_points, registry):
+        """Test loading policies from entry points."""
+        # Mock entry points
+        mock_entry_point = Mock()
+        mock_entry_point.name = "test_policy"
+        mock_entry_point.load.return_value = MinimumNecessaryAccessPolicy
+        
+        mock_entry_points.return_value.select.return_value = [mock_entry_point]
+        
+        # Load from entry points
+        registry.load_policy_from_entry_point("fastmcp.policies")
+        
+        # Check that policy was loaded
+        assert registry.get_policy("test_policy") is not None
+
+
+class TestPolicyIntegration:
+    """Test policy integration with FastMCP server."""
+    
+    @pytest.mark.asyncio
+    async def test_policy_engine_with_server(self):
+        """Test policy engine integration with FastMCP server."""
+        from fastmcp import FastMCP
+        
+        # Create server with policy engine
+        server = FastMCP("Test Server")
+        policy_engine = server.enable_policy_engine()
+        
+        # Register policies
+        policy_engine.register_policy(MinimumNecessaryAccessPolicy())
+        policy_engine.register_policy(RBACPolicy())
+        
+        # Test that policy engine is accessible
+        assert server.get_policy_engine() is not None
+        assert server.get_policy_engine() == policy_engine
+        
+        # Test policy evaluation
+        context = {
+            "user": {"roles": ["user"]},
+            "action": "read",
+            "resource": {"type": "document"}
+        }
+        
+        decision = await policy_engine.evaluate(context)
+        assert decision.allow is True

tests/policy/test_policy_http.py

@@ -0,0 +1,54 @@
+"""Tests for policy HTTP endpoints."""
+
+import pytest
+from fastmcp import FastMCP
+from fastmcp.policy import PolicyEngine
+from fastmcp.policy.policies import MinimumNecessaryAccessPolicy, RBACPolicy
+from starlette.routing import Route
+
+
+class TestPolicyHTTPEndpoint:
+    """Test the policy evaluation HTTP endpoint."""
+    
+    @pytest.fixture
+    def server_with_policy(self):
+        """Create a server with policy engine enabled."""
+        server = FastMCP("Test Policy Server")
+        policy_engine = server.enable_policy_engine()
+        
+        # Register policies
+        policy_engine.register_policy(MinimumNecessaryAccessPolicy())
+        policy_engine.register_policy(RBACPolicy())
+        
+        return server
+    
+    @pytest.fixture
+    def app(self, server_with_policy):
+        """Create the HTTP app with policy endpoint."""
+        return server_with_policy.http_app(transport="sse")
+    
+    def test_policy_evaluate_endpoint_exists(self, app):
+        """Test that the policy evaluation endpoint exists in the app."""
+        # Check that the policy route exists
+        policy_route_found = False
+        for route in app.routes:
+            if isinstance(route, Route) and route.path == "/policy/evaluate":
+                policy_route_found = True
+                break
+        
+        assert policy_route_found, "Policy evaluation endpoint not found in app routes"
+    
+    def test_policy_engine_integration(self, server_with_policy):
+        """Test that policy engine is properly integrated with the server."""
+        # Check that policy engine is enabled
+        assert server_with_policy.get_policy_engine() is not None
+        
+        # Check that policies are registered
+        policy_engine = server_with_policy.get_policy_engine()
+        policies = policy_engine.get_policy_metadata()
+        assert len(policies) == 2
+        
+        # Check that both policies are present
+        policy_names = [p["name"] for p in policies]
+        assert "minimum_necessary_access" in policy_names
+        assert "rbac" in policy_names