Add AgentforceScorer function feature

src/datacustomcode/cli.py

@@ -210,7 +210,11 @@ def deploy(
     sf_cli_org: Optional[str],
     use_in_feature: Optional[str],
 ):
-    from datacustomcode.constants import USE_IN_FEATURE_MAPPING_FOR_CONNECT_API
+    from datacustomcode.constants import (
+        FUNCTION_INVOKE_OPTIONS_FEATURES,
+        REQUEST_TYPE_TO_FEATURE,
+        USE_IN_FEATURE_MAPPING_FOR_CONNECT_API,
+    )
     from datacustomcode.deploy import (
         COMPUTE_TYPES,
         CodeExtensionMetadata,
@@ -252,17 +256,22 @@ def deploy(
         else:
             click.secho(
                 "Error: Function signature does not match a supported type. "
-                "Use SearchIndexChunkingV1Request and "
-                "SearchIndexChunkingV1Response in function signature.",
+                "Use a matching <Feature>V1Request and <Feature>V1Response "
+                "pair in the function signature. Supported types: "
+                + ", ".join(REQUEST_TYPE_TO_FEATURE),
                 fg="red",
             )
             raise click.Abort()
 
-        # Map feature names to Connect API names
-        mapped_feature = USE_IN_FEATURE_MAPPING_FOR_CONNECT_API.get(
-            use_in_feature, use_in_feature
-        )
-        metadata.functionInvokeOptions = [mapped_feature]
+        if use_in_feature in FUNCTION_INVOKE_OPTIONS_FEATURES:
+            # Newer function features only bind through invokeOptions (v67.0)
+            metadata.invokeOptions = [use_in_feature]
+        else:
+            # Map feature names to Connect API names (legacy field)
+            mapped_feature = USE_IN_FEATURE_MAPPING_FOR_CONNECT_API.get(
+                use_in_feature, use_in_feature
+            )
+            metadata.functionInvokeOptions = [mapped_feature]
 
     try:
         if sf_cli_org:
@@ -287,11 +296,14 @@ def deploy(
     default=None,
     help=(
         "Invoke option for this package. For scripts: 'BatchTransform' "
-        "(default) or 'StreamingTransform'. For functions: 'SearchIndexChunking'."
+        "(default) or 'StreamingTransform'. For functions: 'SearchIndexChunking' "
+        "(default) or 'AgentforceScorer'."
     ),
 )
 def init(directory: str, code_type: str, use_in_feature: Optional[str]):
     from datacustomcode.constants import (
+        FEATURE_TEMPLATE_MAPPING,
+        FUNCTION_USE_IN_FEATURE_DEFAULT,
         SCRIPT_USE_IN_FEATURE_BATCH,
         SCRIPT_USE_IN_FEATURE_OPTIONS,
         SCRIPT_USE_IN_FEATURE_STREAMING,
@@ -315,7 +327,14 @@ def init(directory: str, code_type: str, use_in_feature: Optional[str]):
             raise click.Abort()
         streaming = use_in_feature == SCRIPT_USE_IN_FEATURE_STREAMING
     else:
-        use_in_feature = use_in_feature or "SearchIndexChunking"
+        use_in_feature = use_in_feature or FUNCTION_USE_IN_FEATURE_DEFAULT
+        if use_in_feature not in FEATURE_TEMPLATE_MAPPING:
+            click.secho(
+                f"Error: Invalid --use-in-feature '{use_in_feature}' for a "
+                f"function. Valid options: {', '.join(FEATURE_TEMPLATE_MAPPING)}.",
+                fg="red",
+            )
+            raise click.Abort()
 
     click.echo("Copying template to " + click.style(directory, fg="blue", bold=True))
     if code_type == "script":

src/datacustomcode/constants.py

@@ -31,13 +31,23 @@
 # Feature to template folder mapping
 FEATURE_TEMPLATE_MAPPING = {
     "SearchIndexChunking": "chunking",
+    "AgentforceScorer": "agentforce_scorer",
 }
 
-# Feature name to Connect API name mapping
+# Default function feature
+FUNCTION_USE_IN_FEATURE_DEFAULT = "SearchIndexChunking"
+
+# Feature name to Connect API name mapping (legacy functionInvokeOptions field)
 USE_IN_FEATURE_MAPPING_FOR_CONNECT_API = {
     "SearchIndexChunking": "UnstructuredChunking",
 }
 
+# Function features sent via invokeOptions (v67.0) instead of the legacy
+# functionInvokeOptions field
+FUNCTION_INVOKE_OPTIONS_FEATURES = [
+    "AgentforceScorer",
+]
+
 # Script (data transform) invoke options
 SCRIPT_USE_IN_FEATURE_BATCH = "BatchTransform"
 SCRIPT_USE_IN_FEATURE_STREAMING = "StreamingTransform"
@@ -50,4 +60,6 @@
 REQUEST_TYPE_TO_FEATURE = {
     "SearchIndexChunkingV1Request": "SearchIndexChunking",
     "SearchIndexChunkingV1Response": "SearchIndexChunking",
+    "AgentforceScorerV1Request": "AgentforceScorer",
+    "AgentforceScorerV1Response": "AgentforceScorer",
 }

src/datacustomcode/function/feature_types/agentforce_scorer.py

@@ -0,0 +1,207 @@
+# Copyright (c) 2025, Salesforce, Inc.
+# SPDX-License-Identifier: Apache-2
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Pydantic models for Agentforce Scorer V1
+
+The request carries the session tracing detail view of a single Agentforce
+session. The response follows the open-ended scorer result contract
+(``agentforce_session_tracing/scorerOpenEnded.yaml`` in Core).
+
+Field names mirror the wire format (camelCase), so they are not renamed.
+"""
+
+from typing import (
+    List,
+    Optional,
+)
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+)
+
+# Max length of Core "customText" fields
+CUSTOM_TEXT_MAX_LENGTH = 100_000
+
+
+class AgentforceScorerV1SessionState(BaseModel):
+    """Session identifiers and channel information"""
+
+    sessionId: str = Field(description="Agentforce session id")
+    startTimestamp: str = Field(
+        description="Session start (ISO-8601 datetime)",
+        examples=["2026-01-01T10:00:00Z"],
+    )
+    channel: str = Field(description="Channel of the session", examples=["Messaging"])
+    messagingSessionId: Optional[str] = Field(
+        default=None, description="Messaging session id, if any"
+    )
+    voiceCallId: Optional[str] = Field(
+        default=None, description="Voice call id, if any"
+    )
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1Actor(BaseModel):
+    """Participant in the session (end user or agent)"""
+
+    id: str = Field(description="Participant id")
+    participantObject: str = Field(description="Participant object API name")
+    role: str = Field(description="Participant role", examples=["user"])
+    agentApiName: Optional[str] = Field(default=None, description="Agent API name")
+    agentVersion: Optional[str] = Field(default=None, description="Agent version")
+    agentType: Optional[str] = Field(default=None, description="Agent type")
+    agentTemplate: Optional[str] = Field(default=None, description="Agent template")
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1Metrics(BaseModel):
+    """Aggregate session metrics"""
+
+    durationMs: int = Field(description="Session duration in milliseconds")
+    turns: int = Field(description="Number of conversation turns")
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1Message(BaseModel):
+    """Single message exchanged in a run"""
+
+    message: str = Field(description="Message text", max_length=CUSTOM_TEXT_MAX_LENGTH)
+    timestamp: str = Field(
+        description="Message time (ISO-8601 datetime)",
+        examples=["2026-01-01T10:00:05Z"],
+    )
+    actorRole: str = Field(description="Role of the sender", examples=["user"])
+    actorId: str = Field(description="Id of the sender")
+    type: str = Field(description="Message type")
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1Step(BaseModel):
+    """Single step of the agent loop (classification, action, ...)
+
+    ``input``, ``output`` and ``attributes`` are raw strings, often JSON-encoded.
+    """
+
+    stepId: str = Field(description="Step id")
+    type: str = Field(description="Step type")
+    name: str = Field(description="Step name")
+    startTimestamp: str = Field(
+        description="Step start (ISO-8601 datetime)",
+        examples=["2026-01-01T10:00:06Z"],
+    )
+    endTimestamp: str = Field(
+        description="Step end (ISO-8601 datetime)",
+        examples=["2026-01-01T10:00:07Z"],
+    )
+    durationMs: int = Field(description="Step duration in milliseconds")
+    input: Optional[str] = Field(
+        default=None, description="Step input", max_length=CUSTOM_TEXT_MAX_LENGTH
+    )
+    output: Optional[str] = Field(
+        default=None, description="Step output", max_length=CUSTOM_TEXT_MAX_LENGTH
+    )
+    attributes: Optional[str] = Field(
+        default=None,
+        description="Step attributes",
+        max_length=CUSTOM_TEXT_MAX_LENGTH,
+    )
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1AgentLoop(BaseModel):
+    """Agent loop steps executed within a run"""
+
+    steps: List[AgentforceScorerV1Step] = Field(
+        default_factory=list, description="Steps in execution order"
+    )
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1Run(BaseModel):
+    """A run (topic execution) within a session"""
+
+    runId: str = Field(description="Run id")
+    topicName: str = Field(description="Topic handled by this run")
+    startTimestamp: str = Field(
+        description="Run start (ISO-8601 datetime)",
+        examples=["2026-01-01T10:00:00Z"],
+    )
+    endTimestamp: str = Field(
+        description="Run end (ISO-8601 datetime)",
+        examples=["2026-01-01T10:01:00Z"],
+    )
+    durationMs: int = Field(description="Run duration in milliseconds")
+    messages: List[AgentforceScorerV1Message] = Field(
+        default_factory=list, description="Messages in this run"
+    )
+    agentLoop: AgentforceScorerV1AgentLoop = Field(
+        default_factory=AgentforceScorerV1AgentLoop,
+        description="Agent loop of this run",
+    )
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1SessionTracingDetailView(BaseModel):
+    """Full session tracing detail view passed to the scorer"""
+
+    sessionState: AgentforceScorerV1SessionState = Field(
+        description="Session identifiers and channel"
+    )
+    actors: List[AgentforceScorerV1Actor] = Field(
+        default_factory=list, description="Session participants"
+    )
+    metrics: Optional[AgentforceScorerV1Metrics] = Field(
+        default=None, description="Aggregate session metrics"
+    )
+    runs: List[AgentforceScorerV1Run] = Field(
+        default_factory=list, description="Runs in session order"
+    )
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1Output(BaseModel):
+    """Single open-ended scorer result"""
+
+    label: Optional[str] = Field(
+        default=None, description="Result label", examples=["resolution"]
+    )
+    value: Optional[str] = Field(
+        default=None, description="Result value", examples=["resolved"]
+    )
+    isPassed: Optional[bool] = Field(
+        default=None, description="Whether the session passed this check"
+    )
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1Request(BaseModel):
+    """Request for Agentforce Scorer"""
+
+    python_scorer_input: AgentforceScorerV1SessionTracingDetailView = Field(
+        description="Session to score"
+    )
+    model_config = ConfigDict(extra="ignore")
+
+
+class AgentforceScorerV1Response(BaseModel):
+    """Response for Agentforce Scorer (open-ended result)"""
+
+    outputs: List[AgentforceScorerV1Output] = Field(
+        default_factory=list, description="Scorer results"
+    )
+    model_config = ConfigDict(extra="ignore")

src/datacustomcode/templates/function/agentforce_scorer/payload/config.json

@@ -0,0 +1 @@
+{}

src/datacustomcode/templates/function/agentforce_scorer/payload/entrypoint.py

@@ -0,0 +1,43 @@
+import logging
+
+from datacustomcode.function import Runtime
+from datacustomcode.function.feature_types.agentforce_scorer import (
+    AgentforceScorerV1Output,
+    AgentforceScorerV1Request,
+    AgentforceScorerV1Response,
+)
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)
+
+
+def function(
+    request: AgentforceScorerV1Request, runtime: Runtime
+) -> AgentforceScorerV1Response:
+    """Score a single Agentforce session.
+
+    Args:
+        request: AgentforceScorerV1Request with the session tracing detail view
+        runtime: Runtime context (unused but required by contract)
+
+    Returns:
+        AgentforceScorerV1Response with one or more open-ended results
+    """
+    session = request.python_scorer_input
+    messages = [message for run in session.runs for message in run.messages]
+    logger.info(
+        f"Scoring session {session.sessionState.sessionId}: "
+        f"{len(session.runs)} runs, {len(messages)} messages"
+    )
+
+    # Replace with your scoring logic
+    user_turns = sum(1 for message in messages if message.actorRole == "user")
+    return AgentforceScorerV1Response(
+        outputs=[
+            AgentforceScorerV1Output(
+                label="user_turns",
+                value=str(user_turns),
+                isPassed=user_turns > 0,
+            )
+        ]
+    )

src/datacustomcode/templates/function/agentforce_scorer/requirements.txt

@@ -0,0 +1 @@
+# Packages required for the agentforce scorer function