feat: Implement approval mechanism with new error types and AI guidance fields for errors.

Author: tercel

CHANGELOG.md

@@ -5,6 +5,24 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.8.0] - 2026-03-02
+
+### Added
+
+- **Approval system (F-028)**: Full runtime approval support via `ElicitationApprovalHandler` that bridges MCP elicitation to apcore's approval system. New `approval_handler` parameter on `serve()`. Supports `request_approval()` and `check_approval()` methods.
+  - `ElicitationApprovalHandler`: Presents approval requests to users via MCP elicitation. Maps elicit actions (`accept`/`decline`/`cancel`) to `ApprovalResult` statuses.
+  - CLI `--approval` flag with choices: `elicit`, `auto-approve`, `always-deny`, `off` (default).
+- **Approval error codes**: `APPROVAL_DENIED`, `APPROVAL_TIMEOUT`, `APPROVAL_PENDING` added to `ERROR_CODES`.
+- **Enhanced error responses with AI guidance**: `ErrorMapper` now extracts `retryable`, `ai_guidance`, `user_fixable`, and `suggestion` fields from apcore `ModuleError` and includes non-None values in error response dicts. `ExecutionRouter` appends AI guidance as structured JSON to error text content for AI agent consumption.
+- **AI intent metadata in tool descriptions**: `MCPServerFactory.build_tool()` reads `descriptor.metadata` for AI intent keys (`x-when-to-use`, `x-when-not-to-use`, `x-common-mistakes`, `x-workflow-hints`) and appends them to tool descriptions for agent visibility.
+- **Streaming annotation**: `DEFAULT_ANNOTATIONS` now includes `streaming` field. `AnnotationMapper.to_description_suffix()` includes `streaming=true` when the annotation is set.
+
+### Changed
+
+- **`APPROVAL_TIMEOUT` auto-retryable**: `ErrorMapper` sets `retryable=True` for `APPROVAL_TIMEOUT` errors, signaling to AI agents that the operation can be retried.
+- **`APPROVAL_PENDING` includes `approval_id`**: `ErrorMapper` extracts `approval_id` from error details for `APPROVAL_PENDING` errors.
+- **Error text content enriched**: Router error text now includes AI guidance fields as a structured JSON appendix when present, enabling AI agents to parse retry/fix hints.
+
 ## [0.7.0] - 2026-02-28
 
 ### Added
@@ -154,6 +172,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Filtering**: `tags` and `prefix` parameters for selective module exposure.
 - **260 tests**: Unit, integration, E2E, performance, and security test suites.
 
+[0.8.0]: https://github.com/aipartnerup/apcore-mcp-python/compare/v0.7.0...v0.8.0
 [0.7.0]: https://github.com/aipartnerup/apcore-mcp-python/compare/v0.6.0...v0.7.0
 [0.6.0]: https://github.com/aipartnerup/apcore-mcp-python/compare/v0.5.1...v0.6.0
 [0.5.1]: https://github.com/aipartnerup/apcore-mcp-python/compare/v0.5.0...v0.5.1

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "apcore-mcp"
-version = "0.7.0"
+version = "0.8.0"
 description = "Automatic MCP Server & OpenAI Tools Bridge for apcore"
 readme = "README.md"
 license = "Apache-2.0"
@@ -25,7 +25,7 @@ classifiers = [
     "Topic :: Scientific/Engineering :: Artificial Intelligence",
 ]
 dependencies = [
-    "apcore>=0.6.0",
+    "apcore>=0.7.0",
     "mcp>=1.0.0,<2.0",
     "PyJWT>=2.0",
 ]

src/apcore_mcp/__init__.py

@@ -8,6 +8,7 @@
 
 from apcore_mcp._utils import resolve_executor, resolve_registry
 from apcore_mcp.adapters.annotations import AnnotationMapper
+from apcore_mcp.adapters.approval import ElicitationApprovalHandler
 from apcore_mcp.adapters.errors import ErrorMapper
 from apcore_mcp.adapters.id_normalizer import ModuleIDNormalizer
 from apcore_mcp.adapters.schema import SchemaConverter
@@ -39,6 +40,7 @@
     "AuthMiddleware",
     # Adapters
     "AnnotationMapper",
+    "ElicitationApprovalHandler",
     "SchemaConverter",
     "ErrorMapper",
     "ModuleIDNormalizer",
@@ -55,7 +57,7 @@
     "MCP_ELICIT_KEY",
 ]
 
-__version__ = "0.7.0"
+__version__ = "0.8.0"
 
 logger = logging.getLogger(__name__)
 
@@ -82,6 +84,7 @@ def serve(
     authenticator: Authenticator | None = None,
     require_auth: bool = True,
     exempt_paths: set[str] | None = None,
+    approval_handler: object | None = None,
 ) -> None:
     """Launch an MCP Server that exposes all apcore modules as tools.
 
@@ -107,6 +110,7 @@ def serve(
         require_auth: If True, unauthenticated requests receive 401.
             If False, requests proceed without identity (permissive mode).
         exempt_paths: Exact paths that bypass authentication.
+        approval_handler: Optional approval handler for runtime approval support.
     """
     if not name:
         raise ValueError("name must not be empty")
@@ -131,7 +135,7 @@ def serve(
         logging.getLogger("apcore_mcp").setLevel(getattr(logging, log_level.upper()))
 
     registry = resolve_registry(registry_or_executor)
-    executor = resolve_executor(registry_or_executor)
+    executor = resolve_executor(registry_or_executor, approval_handler=approval_handler)
 
     # Build MCP server components
     factory = MCPServerFactory()

src/apcore_mcp/__main__.py

@@ -127,6 +127,17 @@ def _build_parser() -> argparse.ArgumentParser:
         help="Comma-separated paths exempt from auth (default: /health,/metrics).",
     )
 
+    # Approval options
+    parser.add_argument(
+        "--approval",
+        choices=("elicit", "auto-approve", "always-deny", "off"),
+        default="off",
+        help='Approval handler mode (default: "off"). '
+        '"elicit" uses MCP elicitation, '
+        '"auto-approve" auto-approves all requests (dev/testing), '
+        '"always-deny" rejects all requests.',
+    )
+
     return parser
 
 
@@ -219,6 +230,24 @@ def main() -> None:
     if args.exempt_paths:
         exempt_paths_set = set(p.strip() for p in args.exempt_paths.split(","))
 
+    # Build approval handler
+    approval_handler = None
+    if args.approval == "elicit":
+        from apcore_mcp.adapters.approval import ElicitationApprovalHandler
+
+        approval_handler = ElicitationApprovalHandler()
+        logger.info("Approval handler: elicit (MCP elicitation)")
+    elif args.approval == "auto-approve":
+        from apcore.approval import AutoApproveHandler
+
+        approval_handler = AutoApproveHandler()
+        logger.info("Approval handler: auto-approve (dev/testing)")
+    elif args.approval == "always-deny":
+        from apcore.approval import AlwaysDenyHandler
+
+        approval_handler = AlwaysDenyHandler()
+        logger.info("Approval handler: always-deny (enforcement)")
+
     # Launch the MCP server
     try:
         serve(
@@ -234,6 +263,7 @@ def main() -> None:
             authenticator=authenticator,
             require_auth=args.jwt_require_auth,
             exempt_paths=exempt_paths_set,
+            approval_handler=approval_handler,
         )
     except Exception:
         logger.exception("Server startup failed.")

src/apcore_mcp/_utils.py

@@ -14,12 +14,17 @@ def resolve_registry(registry_or_executor: Any) -> Any:
     return registry_or_executor
 
 
-def resolve_executor(registry_or_executor: Any) -> Any:
-    """Get or create an Executor from either a Registry or Executor instance."""
+def resolve_executor(registry_or_executor: Any, *, approval_handler: Any = None) -> Any:
+    """Get or create an Executor from either a Registry or Executor instance.
+
+    Args:
+        registry_or_executor: An apcore Registry or Executor instance.
+        approval_handler: Optional approval handler to pass to new Executor instances.
+    """
     if hasattr(registry_or_executor, "call_async"):
         # Already an Executor
         return registry_or_executor
     # It's a Registry — create a default Executor
     from apcore.executor import Executor
 
-    return Executor(registry_or_executor)
+    return Executor(registry_or_executor, approval_handler=approval_handler)

src/apcore_mcp/adapters/__init__.py

@@ -1,7 +1,8 @@
-"""Adapters: schema conversion, annotation mapping, error mapping, ID normalization."""
+"""Adapters: schema conversion, annotation mapping, error mapping, ID normalization, approval."""
 
 from apcore_mcp.adapters.annotations import AnnotationMapper
+from apcore_mcp.adapters.approval import ElicitationApprovalHandler
 from apcore_mcp.adapters.errors import ErrorMapper
 from apcore_mcp.adapters.id_normalizer import ModuleIDNormalizer
 
-__all__ = ["AnnotationMapper", "ErrorMapper", "ModuleIDNormalizer"]
+__all__ = ["AnnotationMapper", "ElicitationApprovalHandler", "ErrorMapper", "ModuleIDNormalizer"]

src/apcore_mcp/adapters/annotations.py

@@ -10,6 +10,7 @@
     "idempotent": False,
     "requires_approval": False,
     "open_world": True,
+    "streaming": False,
 }
 
 
@@ -83,6 +84,8 @@ def to_description_suffix(self, annotations: Any | None) -> str:
             parts.append(f"requires_approval={str(annotations.requires_approval).lower()}")
         if annotations.open_world != DEFAULT_ANNOTATIONS["open_world"]:
             parts.append(f"open_world={str(annotations.open_world).lower()}")
+        if getattr(annotations, "streaming", False) != DEFAULT_ANNOTATIONS["streaming"]:
+            parts.append(f"streaming={str(getattr(annotations, 'streaming', False)).lower()}")
 
         if not parts:
             return ""

src/apcore_mcp/adapters/approval.py

@@ -0,0 +1,80 @@
+"""ElicitationApprovalHandler: bridges MCP elicitation to apcore's approval system."""
+
+from __future__ import annotations
+
+import logging
+
+from apcore.approval import ApprovalHandler, ApprovalRequest, ApprovalResult
+
+from apcore_mcp.helpers import MCP_ELICIT_KEY
+
+logger = logging.getLogger(__name__)
+
+
+class ElicitationApprovalHandler(ApprovalHandler):
+    """Bridges MCP elicitation to apcore's approval system.
+
+    Uses the MCP elicit callback (injected into Context.data) to present
+    approval requests to the human user via the MCP client.
+    """
+
+    async def request_approval(self, request: ApprovalRequest) -> ApprovalResult:
+        """Request approval via MCP elicitation.
+
+        Extracts the elicit callback from ``request.context.data``, builds
+        an approval message, and maps the elicit response to an
+        ``ApprovalResult``.
+
+        Args:
+            request: The approval request containing module_id, description,
+                arguments, and context.
+
+        Returns:
+            ApprovalResult with status "approved" or "rejected".
+        """
+        # Extract elicit callback from context
+        context = request.context
+        data = getattr(context, "data", None) if context is not None else None
+        if data is None:
+            return ApprovalResult(status="rejected", reason="No context available for elicitation")
+
+        elicit_callback = data.get(MCP_ELICIT_KEY)
+        if elicit_callback is None:
+            return ApprovalResult(status="rejected", reason="No elicitation callback available")
+
+        # Build approval message
+        message = (
+            f"Approval required for tool: {request.module_id}\n\n"
+            f"{request.description}\n\n"
+            f"Arguments: {request.arguments}"
+        )
+
+        try:
+            result = await elicit_callback(message)
+        except Exception:
+            logger.debug("Elicitation approval request failed", exc_info=True)
+            return ApprovalResult(status="rejected", reason="Elicitation request failed")
+
+        if result is None:
+            return ApprovalResult(status="rejected", reason="Elicitation returned no response")
+
+        action = result.get("action") if isinstance(result, dict) else getattr(result, "action", None)
+
+        if action == "accept":
+            return ApprovalResult(status="approved")
+        else:
+            return ApprovalResult(status="rejected", reason=f"User action: {action}")
+
+    async def check_approval(self, approval_id: str) -> ApprovalResult:
+        """Check status of an existing approval.
+
+        Phase B (async polling) is not supported via MCP elicitation since
+        elicitation is stateless.
+
+        Args:
+            approval_id: The approval ID to check.
+
+        Returns:
+            Always returns rejected since Phase B is not supported.
+        """
+        return ApprovalResult(status="rejected", reason="Phase B not supported via MCP elicitation")

src/apcore_mcp/adapters/errors.py

@@ -45,6 +45,16 @@ def to_mcp_error(self, error: Exception) -> dict[str, Any]:
             "details": None,
         }
 
+    # Map apcore ModuleError attribute names (snake_case) to MCP wire format (camelCase).
+    # The wire format uses camelCase to match MCP convention and TypeScript output.
+    # apcore input: error.ai_guidance → MCP output: result["aiGuidance"]
+    _AI_GUIDANCE_FIELDS: dict[str, str] = {
+        "retryable": "retryable",
+        "ai_guidance": "aiGuidance",
+        "user_fixable": "userFixable",
+        "suggestion": "suggestion",
+    }
+
     def _handle_apcore_error(self, error: Exception) -> dict[str, Any]:
         """Handle known apcore errors."""
         code: str = getattr(error, "code", "UNKNOWN")
@@ -73,20 +83,71 @@ def _handle_apcore_error(self, error: Exception) -> dict[str, Any]:
         # Schema validation errors need special formatting
         if code == ERROR_CODES["SCHEMA_VALIDATION_ERROR"] and details is not None:
             formatted_message = self._format_validation_errors(details.get("errors", []))
-            return {
+            result: dict[str, Any] = {
                 "is_error": True,
                 "error_type": code,
                 "message": formatted_message if formatted_message else message,
                 "details": details,
             }
+            self._attach_ai_guidance(error, result)
+            return result
+
+        # Approval errors: pass through with specific handling
+        if code == ERROR_CODES["APPROVAL_PENDING"]:
+            # Narrow details to only approvalId; drop everything else.
+            # apcore uses snake_case (approval_id); output uses camelCase (approvalId) for MCP convention.
+            narrowed = {"approvalId": details["approval_id"]} if details and "approval_id" in details else None
+            result = {
+                "is_error": True,
+                "error_type": code,
+                "message": message,
+                "details": narrowed,
+            }
+            self._attach_ai_guidance(error, result)
+            return result
+
+        if code == ERROR_CODES["APPROVAL_TIMEOUT"]:
+            result = {
+                "is_error": True,
+                "error_type": code,
+                "message": message,
+                "details": details,
+                "retryable": True,
+            }
+            self._attach_ai_guidance(error, result)
+            return result
+
+        if code == ERROR_CODES["APPROVAL_DENIED"]:
+            reason = details.get("reason") if details else None
+            result = {
+                "is_error": True,
+                "error_type": code,
+                "message": message,
+                "details": {"reason": reason} if reason else details,
+            }
+            self._attach_ai_guidance(error, result)
+            return result
 
         # All other apcore errors: pass through message and details
-        return {
+        result = {
             "is_error": True,
             "error_type": code,
             "message": message,
             "details": details,
         }
+        self._attach_ai_guidance(error, result)
+        return result
+
+    def _attach_ai_guidance(self, error: Exception, result: dict[str, Any]) -> None:
+        """Extract AI guidance fields from error and attach non-None values to result.
+
+        Reads snake_case attributes from the apcore error and writes camelCase
+        keys to the MCP result dict (matching MCP/TypeScript convention).
+        """
+        for src_field, dest_field in self._AI_GUIDANCE_FIELDS.items():
+            value = getattr(error, src_field, None)
+            if value is not None and dest_field not in result:
+                result[dest_field] = value
 
     def _format_validation_errors(self, errors: list[dict[str, Any]]) -> str:
         """Format SchemaValidationError field-level errors into readable message."""

src/apcore_mcp/constants.py

@@ -21,6 +21,9 @@
     "MODULE_LOAD_ERROR": "MODULE_LOAD_ERROR",
     "MODULE_EXECUTE_ERROR": "MODULE_EXECUTE_ERROR",
     "GENERAL_INVALID_INPUT": "GENERAL_INVALID_INPUT",
+    "APPROVAL_DENIED": "APPROVAL_DENIED",
+    "APPROVAL_TIMEOUT": "APPROVAL_TIMEOUT",
+    "APPROVAL_PENDING": "APPROVAL_PENDING",
 }
 
 MODULE_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_]*(\.[a-z][a-z0-9_]*)*$")