[api][integration] Support agent skills in python

.licenserc.yaml

@@ -29,6 +29,7 @@ header:
     - '**/*.txt'
     - '**/dependency-reduced-pom.xml'
     - '**/LICENSE.*'
+    - '**/resources/skills/*'
   comment: on-failure
   dependency:
     files:

api/src/main/java/org/apache/flink/agents/api/resource/ResourceType.java

@@ -31,7 +31,8 @@ public enum ResourceType {
     VECTOR_STORE("vector_store"),
     PROMPT("prompt"),
     TOOL("tool"),
-    MCP_SERVER("mcp_server");
+    MCP_SERVER("mcp_server"),
+    SKILL("skill");
 
     private final String value;
 

python/flink_agents/api/agents/react_agent.py

@@ -25,7 +25,11 @@
 
 from flink_agents.api.agents.agent import STRUCTURED_OUTPUT, Agent
 from flink_agents.api.agents.types import OutputSchema
-from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_message import (
+    ChatMessage,
+    MessageRole,
+    find_first_system_message,
+)
 from flink_agents.api.decorators import action
 from flink_agents.api.events.chat_event import ChatRequestEvent, ChatResponseEvent
 from flink_agents.api.events.event import InputEvent, OutputEvent
@@ -185,7 +189,8 @@ def start_action(event: InputEvent, ctx: RunnerContext) -> None:
 
         if schema_prompt:
             instruct = schema_prompt.format_messages()
-            usr_msgs = instruct + usr_msgs
+            index = find_first_system_message(usr_msgs)
+            usr_msgs = usr_msgs[:index + 1] + instruct + usr_msgs[index + 1:]
 
         output_schema = ctx.get_action_config_value(key="output_schema")
 

python/flink_agents/api/agents/tests/test_row_schema.py

@@ -21,7 +21,7 @@
 from flink_agents.api.agents.react_agent import OutputSchema
 
 
-def test_output_schema_serializable() -> None:  # noqa: D103
+def test_output_schema_serializable() -> None:
     schema = OutputSchema(
         output_schema=RowTypeInfo(
             [BasicTypeInfo.INT_TYPE_INFO()],

python/flink_agents/api/chat_message.py

@@ -67,3 +67,11 @@ class ChatMessage(BaseModel):
 
     def __str__(self) -> str:
         return f"{self.role.value}: {self.content}"
+
+
+def find_first_system_message(messages: List[ChatMessage]) -> int:
+    """Helper method to find the index of the first system message."""
+    for i in range(len(messages)):
+        if messages[i].role == MessageRole.SYSTEM:
+            return i
+    return -1

python/flink_agents/api/chat_models/chat_model.py

@@ -22,9 +22,14 @@
 from pydantic import Field
 from typing_extensions import override
 
-from flink_agents.api.chat_message import ChatMessage, MessageRole
+from flink_agents.api.chat_message import (
+    ChatMessage,
+    MessageRole,
+    find_first_system_message,
+)
 from flink_agents.api.prompts.prompt import Prompt
 from flink_agents.api.resource import Resource, ResourceType
+from flink_agents.api.skills import EXECUTE_COMMAND_TOOL, LOAD_SKILL_TOOL
 from flink_agents.api.tools.tool import Tool
 
 
@@ -142,6 +147,8 @@ class BaseChatModelSetup(Resource):
     )
     prompt: Prompt | str | None = None
     tools: List[str] | List[Tool] = Field(default_factory=list)
+    skills: List[str] | None = None
+    skill_discovery_prompt: str | None = None
 
     @property
     @abstractmethod
@@ -158,20 +165,27 @@ def resource_type(cls) -> ResourceType:
     def open(self) -> None:
         self.connection = cast(
             "BaseChatModelConnection",
-            self.get_resource(self.connection, ResourceType.CHAT_MODEL_CONNECTION),
+            self.resource_context.get_resource(self.connection, ResourceType.CHAT_MODEL_CONNECTION),
         )
         if self.prompt is not None:
             if isinstance(self.prompt, str):
                 # Get prompt resource if it's a string
                 self.prompt = cast(
-                    "Prompt", self.get_resource(self.prompt, ResourceType.PROMPT)
+                    "Prompt", self.resource_context.get_resource(self.prompt, ResourceType.PROMPT)
                 )
-        if self.tools is not None:
+        if self.skills is not None:
+            self.skill_discovery_prompt = (
+                self.resource_context.generate_skill_discovery_prompt(*self.skills)
+            )
+            self.tools.extend([LOAD_SKILL_TOOL, EXECUTE_COMMAND_TOOL])
+
+        if len(self.tools) > 0:
             self.tools = [
-                cast("Tool", self.get_resource(tool_name, ResourceType.TOOL))
+                cast("Tool", self.resource_context.get_resource(tool_name, ResourceType.TOOL))
                 for tool_name in self.tools
             ]
 
+
     def chat(self, messages: Sequence[ChatMessage], **kwargs: Any) -> ChatMessage:
         """Execute chat conversation.
 
@@ -211,6 +225,14 @@ def chat(self, messages: Sequence[ChatMessage], **kwargs: Any) -> ChatMessage:
                     prompt_messages.append(msg)
             messages = prompt_messages
 
+        if self.skills is not None:
+            index = find_first_system_message(messages)
+            messages = (
+                messages[: index + 1]
+                + [ChatMessage(role=MessageRole.SYSTEM, content=self.skill_discovery_prompt)]
+                + messages[index + 1 :]
+            )
+
         # Call chat model connection to execute chat
         merged_kwargs = self.model_kwargs.copy()
         merged_kwargs.update(kwargs)

python/flink_agents/api/decorators.py

@@ -189,6 +189,24 @@ def vector_store(func: Callable) -> Callable:
     func._is_vector_store = True
     return func
 
+
+def skills(func: Callable) -> Callable:
+    """Decorator for marking a function declaring skills.
+
+    Parameters
+    ----------
+    func : Callable
+        Function to be decorated.
+
+    Returns:
+    -------
+    Callable
+        Decorator function that marks the target function declare skills.
+    """
+    func._is_skills = True
+    return func
+
+
 def java_resource(cls: Type) -> Type:
     """Decorator to mark a class as Java resource."""
     cls._is_java_resource = True

python/flink_agents/api/embedding_models/embedding_model.py

@@ -93,7 +93,7 @@ def model_kwargs(self) -> Dict[str, Any]:
     def open(self) -> None:
         self.connection = cast(
             "BaseEmbeddingModelConnection",
-            self.get_resource(self.connection, ResourceType.EMBEDDING_MODEL_CONNECTION),
+            self.resource_context.get_resource(self.connection, ResourceType.EMBEDDING_MODEL_CONNECTION),
         )
 
     def _get_connection(self) -> BaseEmbeddingModelConnection:

python/flink_agents/api/memory/tests/test_long_term_memory.py

@@ -23,7 +23,7 @@
 )
 
 
-def test_memory_set_serialization() -> None:  # noqa:D103
+def test_memory_set_serialization() -> None:
     memory_set = MemorySet(
         name="chat_history",
         item_type=ChatMessage,

python/flink_agents/api/resource.py

@@ -18,19 +18,20 @@
 import importlib
 from abc import ABC, abstractmethod
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Callable, Dict, Type
+from typing import TYPE_CHECKING, Any, Dict, Type
 
-from pydantic import BaseModel, Field, PrivateAttr, model_validator
+from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator
+
+from flink_agents.api.resource_context import ResourceContext
 
 if TYPE_CHECKING:
     from flink_agents.api.metric_group import MetricGroup
 
-
 class ResourceType(Enum):
     """Type enum of resource.
 
     Currently, support chat_model, chat_model_server, tool, embedding_model,
-    vector_store and prompt.
+    vector_store, prompt, mcp_server, skill.
     """
 
     CHAT_MODEL = "chat_model"
@@ -41,6 +42,7 @@ class ResourceType(Enum):
     VECTOR_STORE = "vector_store"
     PROMPT = "prompt"
     MCP_SERVER = "mcp_server"
+    SKILL = "skill"
 
 
 class Resource(BaseModel, ABC):
@@ -52,14 +54,13 @@ class Resource(BaseModel, ABC):
 
     Attributes:
     ----------
-    get_resource : Callable[[str, ResourceType], "Resource"]
+    resource_context : ResourceContext
         Get other resource object declared in the same Agent. The first argument is
         resource name and the second argument is resource type.
     """
 
-    get_resource: Callable[[str, ResourceType], "Resource"] = Field(
-        exclude=True, default=None
-    )
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+    resource_context: ResourceContext | None = Field(exclude=True, default=None)
 
     # The metric group bound to this resource, injected in RunnerContext#get_resource
     _metric_group: "MetricGroup | None" = PrivateAttr(default=None)

python/flink_agents/api/resource_context.py

@@ -0,0 +1,42 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you 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.
+################################################################################
+"""Public ResourceContext interface.
+
+Defines the capabilities available to a Resource during execution.
+The concrete implementation lives in :mod:`flink_agents.runtime.resource_context`.
+"""
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from flink_agents.api.resource import Resource, ResourceType
+
+
+class ResourceContext(ABC):
+    """Base abstract class for Resource Context."""
+
+    @abstractmethod
+    def get_resource(self, name: str, resource_type: "ResourceType") -> "Resource":
+        """Get another resource declared in the same Agent."""
+
+    @abstractmethod
+    def generate_skill_discovery_prompt(self, *skill_names: str) -> str:
+        """Generate the skill discovery prompt for the given skill names.
+
+        Returns empty string if no skills are configured.
+        """

python/flink_agents/api/skills.py

@@ -0,0 +1,83 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you 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.
+#################################################################################
+"""Skills configuration resource for agent skills discovery.
+
+Example usage::
+
+    @skills
+    @staticmethod
+    def my_skills() -> Skills:
+        return Skills(paths=["./skills"])
+
+    @skills
+    @staticmethod
+    def combined() -> Skills:
+        return Skills(paths=["./local"], urls=["https://example.com/skills"])
+"""
+from __future__ import annotations
+
+from typing import List
+
+from pydantic import Field
+from typing_extensions import override
+
+from flink_agents.api.resource import ResourceType, SerializableResource
+
+
+class Skills(SerializableResource):
+    """A resource that stores skill location configuration.
+
+    This is the user-facing API for defining where skills are located.
+    The runtime will use this configuration to discover and load skills
+    at initialization time, creating an internal SkillManager.
+
+    Example::
+
+        Skills(paths=["./skills"], allowed_commands=["gh", "git"])
+        Skills(paths=["./local"], urls=["https://example.com/skills"])
+
+    Attributes:
+        paths: List of filesystem paths to load skills from.
+        urls: List of URLs to load skills from.
+        resources: List of Python package resources to load skills from.
+        allowed_script_types: Script types that are allowed to execute
+            from skill directories. Defaults to ``["shell", "python"]``.
+            Supported types: ``"shell"`` (.sh, .bash), ``"python"`` (.py).
+        allowed_commands: Whitelist of executable command names that skills
+            are permitted to run when the command is not a skill script.
+            Only the first token of a command is checked against this list.
+    """
+
+    paths: List[str] = Field(default_factory=list)
+    urls: List[str] = Field(default_factory=list)
+    resources: List[str] = Field(default_factory=list)
+    allowed_script_types: List[str] = Field(
+        default_factory=lambda: ["shell", "python"]
+    )
+    allowed_commands: List[str] = Field(default_factory=list)
+
+    @classmethod
+    @override
+    def resource_type(cls) -> ResourceType:
+        """Return resource type of class."""
+        return ResourceType.SKILL
+
+
+# name of built-in tools needed by using skills
+LOAD_SKILL_TOOL = "load_skill"
+EXECUTE_COMMAND_TOOL = "execute_command"

python/flink_agents/api/tests/test_decorators.py

@@ -24,7 +24,7 @@
 from flink_agents.api.runner_context import RunnerContext
 
 
-def test_action_decorator() -> None:  # noqa D103
+def test_action_decorator() -> None:
     @action(InputEvent)
     def forward_action(event: Event, ctx: RunnerContext) -> None:
         input = event.input
@@ -35,7 +35,7 @@ def forward_action(event: Event, ctx: RunnerContext) -> None:
     assert listen_events == (InputEvent,)
 
 
-def test_action_decorator_listen_multi_events() -> None:  # noqa D103
+def test_action_decorator_listen_multi_events() -> None:
     @action(InputEvent, OutputEvent)
     def forward_action(event: Event, ctx: RunnerContext) -> None:
         input = event.input
@@ -46,7 +46,7 @@ def forward_action(event: Event, ctx: RunnerContext) -> None:
     assert listen_events == (InputEvent, OutputEvent)
 
 
-def test_action_decorator_listen_no_event() -> None:  # noqa D103
+def test_action_decorator_listen_no_event() -> None:
     with pytest.raises(AssertionError):
 
         @action()
@@ -55,7 +55,7 @@ def forward_action(event: Event, ctx: RunnerContext) -> None:
             ctx.send_event(OutputEvent(output=input))
 
 
-def test_action_decorator_listen_non_event_type() -> None:  # noqa D103
+def test_action_decorator_listen_non_event_type() -> None:
     with pytest.raises(AssertionError):
 
         @action(List)