Agent Skills Integration Design

[wenjin272]

Agent Skills Integration

Introduction

Agent Skills are folders of instructions, scripts, and resources that agents can discover and use to do things more accurately and efficiently.

my-skill/
├── SKILL.md          # Required: metadata + instructions
├── scripts/          # Optional: executable code
├── references/       # Optional: documentation
└── assets/           # Optional: templates, resources

Agent Skills can provide the following capabilities for agents:

• For skill authors: Build capabilities once and deploy them across multiple agent products.

• For compatible agents: Support for skills lets end users give agents new capabilities out of the box.

• For teams and enterprises: Capture organizational knowledge in portable, version-controlled packages.

How Agent Skill Works

Skills use progressive disclosure to manage context efficiently:

1. Discovery: At startup, agents load only the name and description of each available skill, just enough to know when it might be relevant.

2. Activation: When a task matches a skill’s description, the agent reads the full SKILL.md instructions into context.

3. Execution: The agent follows the instructions, optionally loading referenced files or executing bundled code as needed.

This means that the Agent Framework needs to provide the following capabilities:

• Provide abstraction and management for Agent Skills, supporting the loading, parsing, and utilization of skills by the Agent.

• Offer built-in tools to support the dynamic loading and execution of skills by the Agent:

  • load_skill: Supports the Agent in loading a complete SKILL.md file.

  •  execute_shell_command: Supports the Agent in executing scripts or commands mentioned in SKILL.md.

  •  load_skill_resource: Supports the Agent in reading other resources required by the skill.

User Interface

How to provide agent skills

Users can provide agent skills for flink-agents jobs in various ways:

• Place the skills in a file system accessible to the job.

  • File systems accessible to the job include:

    • minicluster/standalone: The flink process runs locally. Skills can be placed directly in a specific path on the local file system.

    • k8s: The flink process runs within a pod. Users need to use custom images, pod templates, or pv & pvc mechanisms to embed the skills into the image or mount them into the pod.

  • Folder Format

    opt/
    └── skills/
        └── skill-1/
            └── SKILL.md  
        └── skill-2/
            └── SKILL.md

    At this point, the absolute path of this directory must be specified in the flink-agents job.

    env.add_skills_from_paths(["/opt/skills"])

  • Zip Format: Package the aforementioned skills folder into a zip file; in this case, the absolute path to this zip file must be configured.

    env.add_skills_from_paths(["/opt/skills.zip"])

• Package skills into the job's JAR file.

  • Place agent skills in the job's resources directory.

    flink-agents-job/
    └── src/
        └── main/ 
            └── java/
                └── MyJob.java
            └── resources/
                └── skills/
                    └── skill-1/
                        └── SKILL.md  
                    └── skill-2/
                        └── SKILL.md

    Users can specify the relative path within the resources directory in their assignments.

    env.add_skills_from_resources(["skills"])

  • Similarly, skills can also be packaged as a zip file.

    env.add_skills_from_resources(["skills.zip"])

• Download skills via URL

  • Currently, only supports packaging skills into a zip file and storing them remotely, such as on oss, s3, etc.

    env.add_skills_from_urls(["https://<bucket>.oss-cn-hangzhou.aliyuncs.com/skills.zip"])

  • Future support may include downloading skill folders from FTP links, GitHub links, etc.

How to declare skills to be used

• Job Level: Any Agent and ChatModel are visible in the job

  • AgentExecutionEnvironment: Users can declare the skills to be used via the environment.

    class AgentsExecutionEnvironment(ABC):
        ...
        def add_skills_from_paths(self, paths: List[str]) -> "AgentsExecutionEnvironment":
            """Add skills from paths.
    
            Args:
                paths: List of paths on filesystem for loading skills.
            """
    
        def add_skills_from_resources(
            self, resources: List[str]
        ) -> "AgentsExecutionEnvironment":
            """Add skills from resources in jar.
    
            Args:
                resources: List of resources in the job jar for loading skills.
            """
    
        def add_skills_from_urls(self, urls: List[str]) -> "AgentsExecutionEnvironment":
            """Add skills from urls.
    
            Args:
                urls: List of urls for loading skills, each url should represent a zip file
                      contains one or many skills.
            """

    agents_env = AgentsExecutionEnvironment.get_execution_environment(
            env=env, t_env=t_env
        )
    agents_env.add_skills_from_paths(["/opt/skills"])

  • Configuration：Users can also configure via Config.

    class SkillOptions:
        FROM_PATHS = ConfigOption(
            key="skills.from.paths",
            config_type=list,
            default=None,
        )
        
        FROM_RESOURCES = ConfigOption(
            key="skills.from.resources",
            config_type=list,
            default=None,
        )
        
        FROM_URLS = ConfigOption(
            key="skills.from.urls",
            config_type=list,
            default=None,
        )

    agents_env.get_config().set(SkillOptions.FROM_PATHS,["/opt/skills"])

• Agent Level: Any ChatModel within the Agent is visible.

  • Users declare the skills available to the Agent using the @skillsdecorator.

    class MyAgent(Agent):
        @skills
        @staticmethod
        def my_skills() -> ResourceDescriptor:
            # Users must configure at least one 
            # of the following options: paths, resources, or urls.
            # Supports combined configuration
            return ResourceDescriptor(
                clazz=ResourceName.SKILLS_MANAGER,
                paths=["/path/to/skills"],
                resources=["skills"]
                urls=["https://<bucket>.oss-cn-"
                      "hangzhou.aliyuncs.com/skills.zip"],
            )

• ChatModel Level：

  • Configure the skills available to this Chat Model via skill_names; these must be a subset of the skills declared for the job or agent.

    class MyAgent(Agent):
        @chat_model_setup
        @staticmethod
        def my_chat_model() -> ResourceDescriptor:
            return ResourceDescriptor(
                clazz=ResourceName.ChatModel.OLLAMA_SETUP,
                connection="ollama_connection",
                model=OLLAMA_CHAT_MODEL,
                skill_names=["my_skill_1", "my_skill_2"]
            )

Framework Implementation

Skill Abstraction

AgentSkill is an abstraction of agent skills, containing the following elements:

• name: The name of the skill.

• description: A brief description of the skill.

• instructions: The specific content or steps of the skill.

• resources: Other files used by the skill, such as scripts, templates, and data files.

  class AgentSkill(BaseModel):
      """Representation for agent skill that can be loaded and used by agents.
  
      Attributes
      ----------
      name : str
          The name of the skill.
      description : str
          A brief description of the skill.
      instructions : str
          Detailed instructions of the skill.
      resources : Dict[str, str], optional
          Additional resources required by the skill.
      """
      name: str
      description: str
      instructions: str
      resources: Dict[str, str] = None

Skill Manager

AgentSkillManager is responsible for loading, parsing, and validating Agent Skills.

• Loading

  • Supports loading Agent Skills from the following sources:

    • file system

    • jar packages

    • remote urls

• Parsing

  • Parses file content to construct AgentSkill objects.

  • Supports parsing from zip files.

• Validation

  • Verifies that the AgentSkill does not contain unsupported commands or behaviors.

  class AgentSkillManager(Resource):
      paths: List[str] = Field(default=None)
      resources: List[str] = Filed(default=None)
      urls: List[str] = Field(default=None)
      skills: List[AgentSkill] = Field(default_factory=list)
      
      def load_skills(self) -> None:
          """load and parse skills."""
          # load skills from directory
          # load skills from jar
          # load skills from urls
  
      def get_skills(self) -> List[AgentSkill]:
        """get the registered skills."""
        return self.skills
  
      def get_system_prompt(self, names: List[str] | None = None) -> Prompt
        """provide system prompt that instruct llm when and how to 
        use skills."""
      
      def _load_skills_from_paths(self) -> None:
          """load and parse skills from the paths."""
      
      def _load_skills_from_resources(self) -> None:
          """load and parse skills from jar according to resources."""
        
      def _load_skills_from_urls(self) -> None:
          """fetch and parse skills from the specified URLs."""
          
      def _check_skill(self, skill: AgentSkill) -> None:
          """Check if the skill is supported.
          
          Currently, only shell commands or python scripts within the skill 
          are supported.
          """

Built-in Tools

Built-in tools are the utilities required by the Agent to execute skills during runtime.

• load_skill：load the entire skill.

  def load_skill(manager: AgentSkillManager, name: str) -> str:
      """load the entire skill content."""

• execute_shell_command：execute shell commands, including running python scripts, etc.

  def execute_shell_command(
      command: str,
      timeout: int = 300,
      **kwargs: Any,
  ) -> str:
  """Execute given command and return the return code, standard output and error"""

• load_skill_resource：read the resource files required for the skill, such as reference materials, templates, data, etc.

  def load_skill_resource(manager: AgentSkillManager, name: str, path: str) -> str:
      """load the specific resource."""

Future Works

• Support fetching agent skills from mcp server

  • Currently, FastMCP offers Skills Provider capabilities: https://gofastmcp.com/servers/providers/skills. However, Flink-Agents currently integrates the official SDK. Although FastMCP (22.8k stars) has surpassed the official SDK (21.6k stars) to become the most popular MCP framework in the community, it only provides a Python SDK.

  • Retrieving agent skills via an MCP server can enable dynamic updates of agent skills.

[qingfei1994]

It looks great! I have a few questions:

1. When we add skill files as resources, will they be distributed to TaskManagers in the same way Flink normally distributes JAR files? It would be excellent if we could implement this functionality.

2. Will the skills be cached when they are activated, or will they need to be reloaded every time they are called?

[wenjin272]

Good questions!

When we add skill files as resources, will they be distributed to TaskManagers in the same way Flink normally distributes JAR files? It would be excellent if we could implement this functionality.

The behavior varies depending on the source of the skill files provided by the user:

1. Skill files from the filesystem: Users must place the skill files in a filesystem accessible to the TaskManager (TM); therefore, no distribution is required.

2. Skill files from a JAR: The skill files are distributed together with the JAR file.
3. Skill files from a URL: The ultimate goal is for the JobManager (JM) to download the files from the URL and distribute them similarly to JAR files, potentially leveraging Flink's blob storage. However, in the initial version, this will be implemented as each TM downloading the files independently.

Will the skills be cached when they are activated, or will they need to be reloaded every time they are called?

Yes, the loaded skills will be cached, to avoid reloaded.
If a job utilizes a large number of skills, it may impose significant memory pressure; therefore, a TTL (Time-To-Live) mechanism might be necessary. However, we will not implement this in the initial version.

[qingfei1994]

It looks great, looking forward to this feature, and I'm also willing to help if possible.

[addu390]

Thanks for putting this together @wenjin272!

With skills, would it make sense to introduce a grouping concept (e.g., a "workspace") for bundling skills, tools, etc. so they can be reused across agents?

[wenjin272]

Yes, that is exactly what we aim to achieve. In fact, even before introducing skills, we discussed the need for a catalog-like mechanism to manage resources, including models, tools, MCPs, and skills, even pre-built actions.

As you mentioned, this will enable users to reuse existing Resources and Actions, which we believe is essential for the productization of Flink-Agents. We have already discussed this internally, and I believe we will be able to present our ideas soon. cc @xintongsong

[alnzng]

@wenjin272 Great proposal! The progressive disclosure model, SKILL.md format, and three-level scoping are solid design. The instruction-only approach with 3 built-in tools (load_skill, execute_shell_command, load_skill_resource) matches how Anthropic's official skills work.

After going through the design, I have some thoughts below. Would love to discuss with you and the community.

1. Skill-Tool Binding: Reducing Global Tool Count

Flink Agents supports function calls as tool calls. Each skill can trigger function calls and also execute its bundled scripts. The scripts part is well designed, each skill has its own scripts/ folder, isolated from other skills.

But there is a gap on function calls side. Current design only supports static tool list defined globally on ChatModel (BaseChatModelSetup.tools). Agent will enable all tools in every LLM call. When multiple skills are imported, and each skill needs its own function calls, the global tool list can become large. Research shows this is real problem: reducing tool count improves LLM function-calling accuracy, cost efficiency, and reasoning quality:

• https://blog.langchain.com/react-agent-benchmarking/
• https://www.getmaxim.ai/blog/tool-chaos-no-more-how-were-measuring-model-tool-accuracy-in-the-age-of-mcp/
• https://arxiv.org/abs/2411.15399
• https://vercel.com/blog/we-removed-80-percent-of-our-agents-tools

Suggestion: Add mechanism to define tool list per skill. Model-level tool list only defines global minimum tool set (e.g. the 3 built-in skill tools). When LLM loads a skill, that skill's tools become available. When no skill is active, LLM sees only base tools plus skill catalog in system prompt, enough to discover and load the right skill.

Looking at existing Flink Agents API, each resource type has its own decorator (@chat_model_setup, @tool, @prompt, @mcp_server), each with per-instance configuration. Following same pattern, a @skill (singular) decorator can provide per-skill tool binding:

class MyAgent(Agent):
    @skill
    @staticmethod
    def data_analysis() -> ResourceDescriptor:
        return ResourceDescriptor(
            clazz=ResourceName.SKILL,
            path="/opt/skills/data-analysis",
            tools=["query_api", "generate_report"],  # Tools bound to this skill
        )

    @skill
    @staticmethod
    def monitoring() -> ResourceDescriptor:
        return ResourceDescriptor(
            clazz=ResourceName.SKILL,
            path="/opt/skills/monitoring",
            tools=["check_status"],  # Different tools for this skill
        )

2. Repository Abstraction: Pluggable Skill Sources

The proposed AgentSkillManager handles all source types directly, loading from filesystem paths, JAR resources, and URLs in separate private methods (_load_skills_from_paths(), _load_skills_from_resources(), _load_skills_from_urls()). Adding new skill source like Git repos, databases, or MCP servers requires modifying manager class itself. As skill ecosystem grows and community wants to fetch skills from more sources, this becomes bottleneck.

Suggestion: Extract pluggable AgentSkillRepository interface with built-in implementations for current three sources (filesystem, classpath, URL). AgentSkillManager delegates to repositories instead of handling each source type directly. When @skill decorator specifies a path, it uses FileSystemSkillRepository; a url uses URLSkillRepository, and so on. This also enables MCP-based skill sources mentioned in Future Works: McpSkillRepositorycan plug in without touchingAgentSkillManager`.

[wenjin272]

Thanks for your reply, @alnzng. I believe it is more reasonable to leverage RunnerContext to support dynamic tool loading, as this allows us to decouple tool list management from the chat model. Based on this approach, we can update the tool list when invoking load_skill. I prefer this operation to occur within load_skill itself rather than in ToolCallAction, so that ToolCallAction does not need to be aware of the specific invocation details.

I believe this will refactor the logic for tool registration and tool calls, so it will not be included in the first PR for skills support. However, I recognize that this is indeed important, and I will do my best to incorporate this capability in subsequent PRs for skills support.

[alnzng]

Sounds good @wenjin272 , looking forward to your PRs on this. +1 on using local_skill to update the dynamic tool list.

[wenjin272]

Hi, @alnzng, I have talked about the skill-tool binding with @xintongsong offline. He expressed some concerns: It is beneficial to streamline the context by binding tools to skills and supporting dynamic loading. However, this requires users to explicitly declare the tools bound to each skill, which differs from the behavior of existing agents.

I have investigated LangChain, CrewAI, and DeepAgent, and indeed, none of them currently support binding a tool to a specific skill individually.

• DeepAgent: https://docs.langchain.com/oss/python/deepagents/skills#source-precedence
• CrewAI: https://docs.crewai.com/en/concepts/skills#pattern-1-skills-only-domain-expertise-no-actions-needed

Only LangChain mentioned a similar idea: https://docs.langchain.com/oss/python/langchain/multi-agent/skills#extending-the-pattern, but no specific implementation was provided.

Therefore, I think we can first align with other agent frameworks and support this capability at a more appropriate time in the future. WDYT? @alnzng @xintongsong

[wenjin272]

I see agentscope-java has supported binding Tools to Skills for on-demand activation: https://java.agentscope.io/en/task/agent-skill.html#feature-1-progressive-disclosure-of-tools. And in agentscope(python), I also see a issue wants to implement the same thing: agentscope-ai/agentscope#1055, but it has not been processed yet.

I totally agree that the progressive disclosure of tools can streamline context, thereby improving agent performance. This is also why I initially believed this feature needed to be implemented in version 0.3. However, I believe Xintong's point also makes sense: currently, most agents and agent frameworks do not make this apparent to users. By the way, claude code appears to address this issue through a dedicated tool_search tool.

Therefore, my view remains that we can keep this issue under consideration and implement it at an appropriate or necessary time.

[alnzng]

@wenjin272 I am ok to start with using current model-tools binding mechanism to register tools for all skills.

At least I think we have consensus that loading unnecesary tools could confuse LLM and impact the accuracy of reasonsing output, and ideally we should use progressive disclosure to expose tools on demand. What @xintongsong convinced me is that he was thinking that we can educate the users to use skill's scripts to implement "function calls", basically recommending the users not define skill tools in the global model-tools setup. However, this might not work for the tools exposed by MCP server. I believe the challenages come from how we define tools in current architecture, we can think and discuss more on this when we decided to implement progressive disclosure tool loading.