Files
SergeyMenshykh be8d2619e4 Python: [Breaking] Restructure agent skills to use multi-source architecture (#5584)
* migrate skills to multi source architecture

* Fix ruff lint errors in skills module (ASYNC240, SIM108, E501)

- Use anyio.Path for async file I/O in _FileSkillResource.read()
- Use noqa: ASYNC240 for pure string os.path calls in async context
- Restore pre-commit if/else pattern in InlineSkillScript.run()
- Break long lines to fit 120-char limit in _skills.py and test_skills.py

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: collapse multi-line lambdas to single lines to fix pyright errors

The pyright ignore comments only suppress errors on the same line, so
multi-line lambdas left arguments on continuation lines uncovered.
Collapse both lambdas to single lines matching the existing load_skill
lambda pattern.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: replace untyped lambdas with typed inner functions to fix pyright errors

Python lambdas cannot have type annotations, so pyright reports
reportUnknownLambdaType and reportUnknownArgumentType errors that
cannot be suppressed with inline ignore comments. Replace the
lambdas for read_skill_resource and run_skill_script with typed
inner async functions.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: address PR review feedback on docs and prompt template

- Update with_prompt_template() docstring to document the
  {resource_instructions} placeholder requirement
- Remove stray backslashes after {resource_instructions} and
  {runner_instructions} in DEFAULT_SKILLS_INSTRUCTION_PROMPT
- Update subprocess_script_runner docstring to reflect
  FileSkillScript.full_path usage

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* refactor: replace dict[str, Skill] with Sequence[Skill] in SkillsProvider

Replace internal dict-based skills storage with Sequence[Skill] to
eliminate silent duplicate overwrites and simplify the code. Add
_find_skill helper for case-insensitive linear lookup.

Also fix pyright errors in tests by adding isinstance assertions
before accessing .function on SkillResource/SkillScript base types.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* refactor: add read-time resource path validation in _FileSkillsSource

Move security validation (path-traversal and symlink guards) for
file-based skill resources into _FileSkillsSource, restoring the
read-time checks that existed in main via _read_file_skill_resource.

- Add _get_validated_resource_path static method on _FileSkillsSource
  that validates containment, existence, and symlink safety
- _FileSkillsSource.get_skills() validates resource paths at discovery
  time via _get_validated_resource_path before passing to _FileSkillResource
- Move _normalize_resource_path, _is_path_within_directory, and
  _has_symlink_in_path from module-level into _FileSkillsSource as
  static methods (only used there)
- _FileSkillResource remains a simple path-to-content reader
- Add tests for _get_validated_resource_path security checks

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix: reject str/Path in SkillsProvider constructor to prevent str-as-Sequence ambiguity

Since str is a Sequence, passing a path string to the source parameter
would silently be treated as a sequence of characters instead of a
file source. Add an explicit TypeError with a helpful message pointing
callers to SkillsProvider.from_paths().

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Address PR #5584 review feedback

- Remove .NET reference from _FileSkillResource docstring
- Fix inconsistent resource name example (references/FAQ.md -> references/FAQ)
- Simplify SkillsProvider usage in code_defined_skill sample (pass single skill directly)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* remove skillsproviderbuilder

* Update python/packages/core/agent_framework/_skills.py

Co-authored-by: Eduard van Valkenburg <eavanvalkenburg@users.noreply.github.com>

* fix: remove dead code and fix sync function call in InlineSkillResource.read()

- Change await self.function() to self.function() for sync functions
  without **kwargs; async results are handled by inspect.isawaitable()
- Remove unreachable raise ValueError since __init__ already validates

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* remove full_path unnecessary property

* replace anyio with asyncio.to_thread for file I/O in _FileSkillResource

Replace anyio.Path usage with asyncio.to_thread + pathlib.Path since
anyio is not a direct dependency of core (transitive via mcp).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* simplify awaitable check to return directly

Use 'return await result' instead of assigning then returning.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* address PR review feedback for skills refactoring

- Replace anyio with asyncio.to_thread + pathlib.Path for file I/O
- Simplify awaitable check to return directly
- Remove unnecessary function None guard in InlineSkillResource.read()
- Add assert for type narrowing on self.function

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* address PR review feedback for skills refactoring

- Replace anyio with asyncio.to_thread + pathlib.Path for file I/O
- Simplify awaitable checks to return directly
- Remove unnecessary function None guard in InlineSkillResource.read()
- Use typing.cast instead of assert for type narrowing
- Add caching behavior note to SkillsProvider docstring

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* refactor: move name/description from abstract properties to Skill.__init__

Replace abstract properties for name and description on the Skill ABC
with a base __init__ that validates and stores them as regular
attributes. This simplifies custom Skill subclasses (only content
remains abstract) and centralizes validation in the base class,
consistent with SkillResource and SkillScript base classes.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Eduard van Valkenburg <eavanvalkenburg@users.noreply.github.com>
2026-05-06 09:45:06 +00:00

130 lines
4.6 KiB
Python

# Copyright (c) Microsoft. All rights reserved.
import asyncio
import os
# Uncomment this filter to suppress the experimental Skills warning before
# using the sample's Skills APIs.
# import warnings
# warnings.filterwarnings("ignore", message=r"\[SKILLS\].*", category=FutureWarning)
from textwrap import dedent
from agent_framework import Agent, InlineSkill, SkillsProvider
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
from dotenv import load_dotenv
"""
Skill Script Approval — Require human approval before executing skill scripts
This sample demonstrates how to use ``require_script_approval=True`` on
:class:`SkillsProvider` so that every call to ``run_skill_script`` is
gated by a human-in-the-loop approval step.
How it works:
1. A code-defined skill with a script is registered via SkillsProvider.
2. ``require_script_approval=True`` causes the agent to pause and return
approval requests in ``result.user_input_requests`` instead of executing
scripts immediately.
3. The application inspects each request and calls
``request.to_function_approval_response(approved=True|False)`` to approve
or reject.
4. The approval response is sent back via ``agent.run(approval_response, session=session)``
and the agent continues — executing the script if approved, or receiving
an error if rejected.
Prerequisites:
- FOUNDRY_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint.
- FOUNDRY_MODEL (defaults to "gpt-4o-mini").
"""
# Load environment variables from .env file
load_dotenv()
# Define a code skill with a script that performs a sensitive operation
deployment_skill = InlineSkill(
name="deployment",
description="Tools for deploying application versions to production",
instructions=dedent("""\
Use this skill when the user asks to deploy an application.
1. Run the deploy script with the version and environment parameters.
"""),
)
@deployment_skill.script
def deploy(version: str, environment: str = "staging") -> str:
"""Deploy the application to the specified environment."""
return f"Deployed version {version} to {environment}"
async def main() -> None:
"""Run the skill script approval demo."""
endpoint = os.environ["FOUNDRY_PROJECT_ENDPOINT"]
deployment = os.environ.get("FOUNDRY_MODEL", "gpt-4o-mini")
client = FoundryChatClient(
project_endpoint=endpoint,
model=deployment,
credential=AzureCliCredential(),
)
# Create the skills provider with script approval enabled
skills_provider = SkillsProvider(
source=[deployment_skill],
require_script_approval=True,
)
async with Agent(
client=client,
instructions="You are a deployment assistant. Use the deployment skill to deploy applications.",
context_providers=[skills_provider],
) as agent:
session = agent.create_session()
print("Starting agent with skill script approval enabled...")
print("-" * 60)
# Step 1: Send the user request — the agent will try to call the script
query = "Deploy the latest application version 2.5.0 to the production environment"
print(f"User: {query}")
result = await agent.run(query, session=session)
# Step 2: Handle approval requests (with sessions, context is
# maintained automatically — just send the approval response)
while result.user_input_requests:
for request in result.user_input_requests:
print("\nApproval needed:")
print(f" Function: {request.function_call.name}") # type: ignore[union-attr]
print(f" Arguments: {request.function_call.arguments}") # type: ignore[union-attr]
# In a real application, prompt the user here
approved = True # Change to False to see rejection
print(f" Decision: {'Approved' if approved else 'Rejected'}")
# Send the approval response — session preserves conversation history
approval_response = request.to_function_approval_response(approved=approved)
result = await agent.run(approval_response, session=session)
print(f"\nAgent: {result}")
if __name__ == "__main__":
asyncio.run(main())
"""
Sample output:
Starting agent with skill script approval enabled...
------------------------------------------------------------
User: Deploy version 2.5.0 to production
Approval needed:
Function: run_skill_script
Arguments: {"skill_name": "deployment", "script_name": "deploy", ...}
Decision: Approved
Agent: Successfully deployed version 2.5.0 to production.
"""