diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index ef03652898..358f7141cb 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -59,7 +59,7 @@ register_state_type, ) from ._settings import SecretString, load_settings -from ._skills import Skill, SkillResource, SkillsProvider +from ._skills import Skill, SkillContext, SkillResource, SkillsProvider from ._telemetry import ( AGENT_FRAMEWORK_USER_AGENT, APP_INFO, @@ -270,6 +270,7 @@ "SessionContext", "SingleEdgeGroup", "Skill", + "SkillContext", "SkillResource", "SkillsProvider", "SubWorkflowRequestMessage", diff --git a/python/packages/core/agent_framework/_skills.py b/python/packages/core/agent_framework/_skills.py index fc71329a5f..5269347811 100644 --- a/python/packages/core/agent_framework/_skills.py +++ b/python/packages/core/agent_framework/_skills.py @@ -29,10 +29,14 @@ import logging import os import re -from collections.abc import Callable, Sequence +import typing +from collections.abc import Awaitable, Callable, Sequence +from dataclasses import dataclass from html import escape as xml_escape from pathlib import Path, PurePosixPath -from typing import TYPE_CHECKING, Any, ClassVar, Final +from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Union, get_origin, overload + +from typing_extensions import TypeVar from ._sessions import BaseContextProvider from ._tools import FunctionTool @@ -45,6 +49,105 @@ # region Models +DepsT = TypeVar("DepsT", default=None) + + +@dataclass +class SkillContext(Generic[DepsT]): + """Typed context provided to skill resource functions. + + .. warning:: Experimental + + This API is experimental and subject to change or removal + in future versions without notice. + + A generic context object that carries typed dependencies into resource + functions. Resource functions that declare a ``SkillContext[DepsT]`` + first parameter receive this context automatically when invoked. + + The context is mutable so that resources can enrich :attr:`deps` with + new state for subsequent resources (e.g. one resource loads data into + ``deps.db``, and a later resource queries it). + + Attributes: + deps: The dependency object supplied to :class:`Skill`. + + Examples: + .. code-block:: python + + from dataclasses import dataclass + from agent_framework import Skill, SkillContext, SkillsProvider + + + @dataclass + class MyDeps: + db: DatabaseClient + api_key: str + + + skill = Skill( + name="my-skill", + description="...", + content="...", + deps=MyDeps(db=conn, api_key="..."), + ) + + + @skill.resource + async def get_data(ctx: SkillContext[MyDeps]) -> str: + result = await ctx.deps.db.query("SELECT ...") + return str(result) + + + provider = SkillsProvider(skills=[skill]) + """ + + deps: DepsT + + +# Union of all accepted resource function signatures. +# Used by the ``@Skill.resource`` overloads to constrain decorated callables. +ResourceFunc = Union[ + Callable[[SkillContext[DepsT]], str], + Callable[[SkillContext[DepsT]], Awaitable[str]], + Callable[[], str], + Callable[[], Awaitable[str]], +] + + +def _is_skill_ctx(annotation: Any) -> bool: + """Return whether *annotation* is the ``SkillContext`` class, parameterized or not.""" + return annotation is SkillContext or get_origin(annotation) is SkillContext + + +def _resolve_takes_ctx(func: Callable[..., Any]) -> bool: + """Return whether *func*'s first positional parameter is :class:`SkillContext`. + + Uses :func:`typing.get_type_hints` to resolve annotations, including + stringified forms produced by ``from __future__ import annotations``. + + .. note:: + + Deps classes must be defined at **module level** so that + :func:`typing.get_type_hints` can resolve them. + """ + sig = inspect.signature(func) + first = next( + (p for p in sig.parameters.values() if p.kind in (_POS_ONLY, _POS_OR_KW)), + None, + ) + if first is None or first.annotation is inspect.Parameter.empty: + return False + try: + resolved = typing.get_type_hints(func).get(first.name) + return resolved is not None and _is_skill_ctx(resolved) + except Exception: + return False + + +_POS_ONLY = inspect.Parameter.POSITIONAL_ONLY +_POS_OR_KW = inspect.Parameter.POSITIONAL_OR_KEYWORD + class SkillResource: """A named piece of supplementary content attached to a skill. @@ -107,6 +210,10 @@ def __init__( self.content = content self.function = function + # Precompute whether the first positional parameter is typed as + # SkillContext to avoid repeated inspection at invocation time. + self._takes_ctx: bool = _resolve_takes_ctx(function) if function is not None else False + # Precompute whether the function accepts **kwargs to avoid # repeated inspect.signature() calls on every invocation. self._accepts_kwargs: bool = False @@ -115,7 +222,7 @@ def __init__( self._accepts_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()) -class Skill: +class Skill(Generic[DepsT]): """A skill definition with optional resources. .. warning:: Experimental @@ -128,6 +235,10 @@ class Skill: supplied at construction time or added later via the :meth:`resource` decorator. + When ``deps`` is provided, the class is parameterized on the + dependency type so that Pyright/mypy can verify that ``@skill.resource`` + functions declare a matching ``SkillContext[DepsT]`` parameter. + Attributes: name: Skill name (lowercase letters, numbers, hyphens only). description: Human-readable description of the skill. @@ -135,6 +246,7 @@ class Skill: resources: Mutable list of :class:`SkillResource` instances. path: Absolute path to the skill directory on disk, or ``None`` for code-defined skills. + deps: The dependency instance, or ``None``. Examples: Direct construction: @@ -148,20 +260,31 @@ class Skill: resources=[SkillResource(name="ref", content="...")], ) - With dynamic resources: + With typed dependencies: .. code-block:: python + @dataclass + class MyDeps: + db: DatabaseClient + + skill = Skill( name="db-skill", description="Database operations", content="Use this skill for DB tasks.", + deps=MyDeps(db=conn), ) @skill.resource - def get_schema() -> str: - return "CREATE TABLE ..." + async def get_schema(ctx: SkillContext[MyDeps]) -> str: + return str(await ctx.deps.db.query("SHOW TABLES")) + + + @skill.resource + def get_version() -> str: + return "1.0" """ def __init__( @@ -172,6 +295,7 @@ def __init__( content: str, resources: list[SkillResource] | None = None, path: str | None = None, + deps: DepsT | None = None, ) -> None: """Initialize a Skill. @@ -182,6 +306,11 @@ def __init__( resources: Pre-built resources to attach to this skill. path: Absolute path to the skill directory on disk. Set automatically for file-based skills; leave as ``None`` for code-defined skills. + deps: The dependency instance for this skill. When provided, + Pyright/mypy infer the generic type parameter and verify that + ``@skill.resource`` callables declare a matching + ``SkillContext[DepsT]`` first parameter. Passed to resource + functions at invocation time. """ if not name or not name.strip(): raise ValueError("Skill name cannot be empty.") @@ -193,6 +322,34 @@ def __init__( self.content = content self.resources: list[SkillResource] = resources if resources is not None else [] self.path = path + self.deps = deps + + # --- Overloads: one per accepted resource function signature --- + + @overload + def resource(self, func: Callable[[SkillContext[DepsT]], str], /) -> Callable[[SkillContext[DepsT]], str]: ... + + @overload + def resource( + self, func: Callable[[SkillContext[DepsT]], Awaitable[str]], / + ) -> Callable[[SkillContext[DepsT]], Awaitable[str]]: ... + + @overload + def resource(self, func: Callable[[], str], /) -> Callable[[], str]: ... + + @overload + def resource(self, func: Callable[[], Awaitable[str]], /) -> Callable[[], Awaitable[str]]: ... + + @overload + def resource( + self, + func: None = None, + *, + name: str | None = None, + description: str | None = None, + ) -> Callable[[ResourceFunc[DepsT]], ResourceFunc[DepsT]]: ... + + # --- Implementation --- def resource( self, @@ -370,6 +527,33 @@ class SkillsProvider(BaseContextProvider): skills=[my_skill], ) + With typed dependencies: + + .. code-block:: python + + from dataclasses import dataclass + + + @dataclass + class MyDeps: + db: DatabaseClient + + + skill = Skill( + name="db-skill", + description="DB operations", + content="...", + deps=MyDeps(db=conn), + ) + + + @skill.resource + async def get_data(ctx: SkillContext[MyDeps]) -> str: + return str(await ctx.deps.db.query("SELECT ...")) + + + provider = SkillsProvider(skills=[skill]) + Attributes: DEFAULT_SOURCE_ID: Default value for the ``source_id`` used by this provider. """ @@ -380,7 +564,7 @@ def __init__( self, skill_paths: str | Path | Sequence[str | Path] | None = None, *, - skills: Sequence[Skill] | None = None, + skills: Sequence[Skill[Any]] | None = None, instruction_template: str | None = None, resource_extensions: tuple[str, ...] | None = None, source_id: str | None = None, @@ -523,7 +707,9 @@ async def _read_skill_resource(self, skill_name: str, resource_name: str, **kwar Resolves the resource by case-insensitive name lookup. Static ``content`` is returned directly; callable resources are invoked - (awaited if async). + (awaited if async). Resource functions that declare a + :class:`SkillContext` first parameter receive a context carrying + the skill's ``deps``. Args: skill_name: The name of the owning skill. @@ -559,12 +745,18 @@ async def _read_skill_resource(self, skill_name: str, resource_name: str, **kwar if resource.function is not None: try: + # Build positional args: prepend SkillContext if the resource expects it. + args: tuple[Any, ...] = () + if resource._takes_ctx: # pyright: ignore[reportPrivateUsage] + args = (SkillContext(deps=skill.deps),) + + # Forward **kwargs to resource functions that accept them. + fwd_kwargs: dict[str, Any] = kwargs if resource._accepts_kwargs else {} # pyright: ignore[reportPrivateUsage] + if inspect.iscoroutinefunction(resource.function): - result = ( - await resource.function(**kwargs) if resource._accepts_kwargs else await resource.function() # pyright: ignore[reportPrivateUsage] - ) + result = await resource.function(*args, **fwd_kwargs) else: - result = resource.function(**kwargs) if resource._accepts_kwargs else resource.function() # pyright: ignore[reportPrivateUsage] + result = resource.function(*args, **fwd_kwargs) return str(result) except Exception as exc: logger.exception("Failed to read resource '%s' from skill '%s'", resource_name, skill_name) @@ -858,7 +1050,7 @@ def _search(directory: str, current_depth: int) -> None: return discovered -def _read_file_skill_resource(skill: Skill, resource_name: str) -> str: +def _read_file_skill_resource(skill: Skill[Any], resource_name: str) -> str: """Read a file-based resource from disk with security guards. Validates that the resolved path stays within the skill directory and @@ -902,7 +1094,7 @@ def _read_file_skill_resource(skill: Skill, resource_name: str) -> str: def _discover_file_skills( skill_paths: str | Path | Sequence[str | Path] | None, resource_extensions: tuple[str, ...] = DEFAULT_RESOURCE_EXTENSIONS, -) -> dict[str, Skill]: +) -> dict[str, Skill[Any]]: """Discover, parse, and load all file-based skills from the given paths. Each discovered ``SKILL.md`` is parsed for metadata, and resource files @@ -923,7 +1115,7 @@ def _discover_file_skills( [str(skill_paths)] if isinstance(skill_paths, (str, Path)) else [str(p) for p in skill_paths] ) - skills: dict[str, Skill] = {} + skills: dict[str, Skill[Any]] = {} discovered = _discover_skill_directories(resolved_paths) logger.info("Discovered %d potential skills", len(discovered)) @@ -943,7 +1135,7 @@ def _discover_file_skills( ) continue - file_skill = Skill( + file_skill: Skill[Any] = Skill( name=name, description=description, content=content, @@ -964,9 +1156,9 @@ def _discover_file_skills( def _load_skills( skill_paths: str | Path | Sequence[str | Path] | None, - skills: Sequence[Skill] | None, + skills: Sequence[Skill[Any]] | None, resource_extensions: tuple[str, ...], -) -> dict[str, Skill]: +) -> dict[str, Skill[Any]]: """Discover and merge skills from file paths and code-defined skills. File-based skills are discovered first. Code-defined skills are then @@ -1019,7 +1211,7 @@ def _create_resource_element(resource: SkillResource) -> str: def _create_instructions( prompt_template: str | None, - skills: dict[str, Skill], + skills: dict[str, Skill[Any]], ) -> str | None: """Create the system-prompt text that advertises available skills. diff --git a/python/packages/core/tests/core/test_skills.py b/python/packages/core/tests/core/test_skills.py index cb829b7b9f..3ab92353f2 100644 --- a/python/packages/core/tests/core/test_skills.py +++ b/python/packages/core/tests/core/test_skills.py @@ -11,7 +11,7 @@ import pytest -from agent_framework import SessionContext, Skill, SkillResource, SkillsProvider +from agent_framework import SessionContext, Skill, SkillContext, SkillResource, SkillsProvider from agent_framework._skills import ( DEFAULT_RESOURCE_EXTENSIONS, _create_instructions, @@ -25,10 +25,30 @@ _normalize_resource_path, _read_and_parse_skill_file, _read_file_skill_resource, + _resolve_takes_ctx, _validate_skill_metadata, ) +# Module-level deps classes for SkillContext tests. +# Must be at module level so typing.get_type_hints() can resolve them +# when ``from __future__ import annotations`` is active. +class _SyncDeps: + value = "hello" + + +class _AsyncDeps: + data = "async-data" + + +class _MutableDeps: + loaded: bool = False + + +class _EmptyDeps: + pass + + def _symlinks_supported(tmp: Path) -> bool: """Return True if the current platform/environment supports symlinks.""" test_target = tmp / "_symlink_test_target" @@ -994,41 +1014,279 @@ async def test_read_unknown_resource_returns_error(self) -> None: result = await provider._read_skill_resource("prog-skill", "nonexistent") assert result.startswith("Error:") - async def test_read_callable_resource_sync_with_kwargs(self) -> None: + async def test_read_resource_sync_with_skill_context(self) -> None: + """Sync resource receiving SkillContext gets typed deps.""" + skill = Skill(name="prog-skill", description="A skill.", content="Body", deps=_SyncDeps()) + + @skill.resource + def get_info(ctx: SkillContext[_SyncDeps]) -> str: + return f"info: {ctx.deps.value}" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "get_info") + assert result == "info: hello" + + async def test_read_resource_async_with_skill_context(self) -> None: + """Async resource receiving SkillContext gets typed deps.""" + skill = Skill(name="prog-skill", description="A skill.", content="Body", deps=_AsyncDeps()) + + @skill.resource + async def get_data(ctx: SkillContext[_AsyncDeps]) -> str: + return f"result: {ctx.deps.data}" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "get_data") + assert result == "result: async-data" + + async def test_read_resource_without_context_ignores_deps(self) -> None: + """Resources without SkillContext still work when skill has deps.""" + skill = Skill(name="prog-skill", description="A skill.", content="Body", deps={"ignored": True}) + + @skill.resource + def plain_resource() -> str: + return "plain" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "plain_resource") + assert result == "plain" + + async def test_read_resource_skill_context_with_none_deps(self) -> None: + """SkillContext works when deps is None (default).""" + skill = Skill(name="prog-skill", description="A skill.", content="Body") + + @skill.resource + def get_deps_info(ctx: SkillContext[None]) -> str: + return f"deps={ctx.deps}" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "get_deps_info") + assert result == "deps=None" + + async def test_read_resource_skill_context_mutates_deps(self) -> None: + """Resource can mutate deps for use by subsequent resource calls.""" + deps = _MutableDeps() + skill = Skill(name="prog-skill", description="A skill.", content="Body", deps=deps) + + @skill.resource + def load_data(ctx: SkillContext[_MutableDeps]) -> str: + ctx.deps.loaded = True + return "loaded" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "load_data") + assert result == "loaded" + assert deps.loaded is True + + async def test_read_resource_sync_with_kwargs(self) -> None: + """Sync resource with **kwargs receives forwarded kwargs.""" skill = Skill(name="prog-skill", description="A skill.", content="Body") @skill.resource - def get_user_config(**kwargs: Any) -> str: - user_id = kwargs.get("user_id", "unknown") - return f"config for {user_id}" + def get_info(**kwargs: Any) -> str: + return f"user={kwargs.get('user_id')}" provider = SkillsProvider(skills=[skill]) - result = await provider._read_skill_resource("prog-skill", "get_user_config", user_id="user_123") - assert result == "config for user_123" + result = await provider._read_skill_resource("prog-skill", "get_info", user_id="alice") + assert result == "user=alice" - async def test_read_callable_resource_async_with_kwargs(self) -> None: + async def test_read_resource_async_with_kwargs(self) -> None: + """Async resource with **kwargs receives forwarded kwargs.""" skill = Skill(name="prog-skill", description="A skill.", content="Body") @skill.resource - async def get_user_data(**kwargs: Any) -> str: - token = kwargs.get("auth_token", "none") - return f"data with token={token}" + async def get_data(**kwargs: Any) -> str: + return f"req={kwargs.get('request_id')}" provider = SkillsProvider(skills=[skill]) - result = await provider._read_skill_resource("prog-skill", "get_user_data", auth_token="abc") - assert result == "data with token=abc" + result = await provider._read_skill_resource("prog-skill", "get_data", request_id="42") + assert result == "req=42" - async def test_read_callable_resource_without_kwargs_ignores_extra_args(self) -> None: - """Resource functions without **kwargs should still work when kwargs are passed.""" + async def test_read_resource_kwargs_not_forwarded_when_not_accepted(self) -> None: + """Resources without **kwargs don't receive forwarded kwargs.""" skill = Skill(name="prog-skill", description="A skill.", content="Body") @skill.resource - def static_resource() -> str: - return "static content" + def plain() -> str: + return "ok" provider = SkillsProvider(skills=[skill]) - result = await provider._read_skill_resource("prog-skill", "static_resource", user_id="ignored") - assert result == "static content" + # Extra kwargs passed but function doesn't accept them; should not fail. + result = await provider._read_skill_resource("prog-skill", "plain", user_id="alice") + assert result == "ok" + + async def test_read_resource_ctx_and_kwargs_combined(self) -> None: + """Resource with both SkillContext and **kwargs gets both.""" + skill = Skill(name="prog-skill", description="A skill.", content="Body", deps=_SyncDeps()) + + @skill.resource + def get_combo(ctx: SkillContext[_SyncDeps], **kwargs: Any) -> str: + return f"deps={ctx.deps.value},user={kwargs.get('user_id')}" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "get_combo", user_id="bob") + assert result == "deps=hello,user=bob" + + async def test_read_resource_async_ctx_and_kwargs_combined(self) -> None: + """Async resource with both SkillContext and **kwargs gets both.""" + skill = Skill(name="prog-skill", description="A skill.", content="Body", deps=_AsyncDeps()) + + @skill.resource + async def get_async_combo(ctx: SkillContext[_AsyncDeps], **kwargs: Any) -> str: + return f"data={ctx.deps.data},req={kwargs.get('request_id')}" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "get_async_combo", request_id="99") + assert result == "data=async-data,req=99" + + async def test_read_resource_kwargs_empty_when_none_passed(self) -> None: + """Resource with **kwargs receives empty dict when no extra kwargs passed.""" + skill = Skill(name="prog-skill", description="A skill.", content="Body") + + @skill.resource + def get_info(**kwargs: Any) -> str: + return f"count={len(kwargs)}" + + provider = SkillsProvider(skills=[skill]) + result = await provider._read_skill_resource("prog-skill", "get_info") + assert result == "count=0" + + def test_skill_deps_attribute(self) -> None: + """Skill.deps stores the provided dependency instance.""" + deps = _SyncDeps() + skill = Skill(name="prog-skill", description="A skill.", content="Body", deps=deps) + assert skill.deps is deps + + def test_skill_deps_defaults_to_none(self) -> None: + """Skill.deps defaults to None when not provided.""" + skill = Skill(name="prog-skill", description="A skill.", content="Body") + assert skill.deps is None + + def test_takes_ctx_false_for_no_params(self) -> None: + """_takes_ctx is False when function has no parameters.""" + + def no_params() -> str: + return "" + + resource = SkillResource(name="r", function=no_params) + assert resource._takes_ctx is False + + def test_takes_ctx_false_for_wrong_annotation(self) -> None: + """_takes_ctx is False when first param is not SkillContext.""" + + def wrong_type(x: str) -> str: + return x + + resource = SkillResource(name="r", function=wrong_type) + assert resource._takes_ctx is False + + def test_takes_ctx_false_for_no_annotation(self) -> None: + """_takes_ctx is False for unannotated first parameter.""" + + def no_annotation(x) -> str: # noqa: ANN001 + return str(x) + + resource = SkillResource(name="r", function=no_annotation) + assert resource._takes_ctx is False + + def test_takes_ctx_true_for_bare_skill_context(self) -> None: + """_takes_ctx is True for bare SkillContext (no type param).""" + + def with_ctx(ctx: SkillContext) -> str: # type: ignore[type-arg] + return "" + + resource = SkillResource(name="r", function=with_ctx) + assert resource._takes_ctx is True + + def test_accepts_kwargs_true(self) -> None: + """_accepts_kwargs is True when function has **kwargs.""" + + def with_kwargs(**kwargs: Any) -> str: + return "" + + resource = SkillResource(name="r", function=with_kwargs) + assert resource._accepts_kwargs is True + + def test_accepts_kwargs_false_for_no_kwargs(self) -> None: + """_accepts_kwargs is False when function has no **kwargs.""" + + def no_kwargs() -> str: + return "" + + resource = SkillResource(name="r", function=no_kwargs) + assert resource._accepts_kwargs is False + + def test_accepts_kwargs_true_with_ctx(self) -> None: + """_accepts_kwargs is True when function has both SkillContext and **kwargs.""" + + def with_ctx_and_kwargs(ctx: SkillContext[_SyncDeps], **kwargs: Any) -> str: + return "" + + resource = SkillResource(name="r", function=with_ctx_and_kwargs) + assert resource._takes_ctx is True + assert resource._accepts_kwargs is True + + def test_accepts_kwargs_false_for_static_content(self) -> None: + """_accepts_kwargs is False for static-content resources.""" + resource = SkillResource(name="r", content="static") + assert resource._accepts_kwargs is False + + +class TestResolveSkillContextAnnotationForms: + """Tests for _resolve_takes_ctx covering all annotation forms. + + Because this test file uses ``from __future__ import annotations``, + inline annotations are strings and exercise the ``get_type_hints()`` + resolution path. + """ + + # -- Forms resolved via get_type_hints() (stringified by __future__) -- + + def test_parameterized_skill_context_via_future_annotations(self) -> None: + """SkillContext[T] with from __future__ import annotations.""" + + def func(ctx: SkillContext[_EmptyDeps]) -> str: + return "" + + assert _resolve_takes_ctx(func) is True + + def test_bare_skill_context_via_future_annotations(self) -> None: + """Bare SkillContext with from __future__ import annotations.""" + + def func(ctx: SkillContext) -> str: # type: ignore[type-arg] + return "" + + assert _resolve_takes_ctx(func) is True + + # -- Negative cases -- + + def test_unresolvable_string_annotation_returns_false(self) -> None: + """String annotation that get_type_hints can't resolve returns False.""" + + def func(ctx) -> str: # noqa: ANN001 + return "" + + func.__annotations__["ctx"] = "SomeOtherContext[Deps]" + assert _resolve_takes_ctx(func) is False + + def test_no_positional_params(self) -> None: + """Function with only **kwargs returns False.""" + + def func(**kwargs: Any) -> str: + return "" + + assert _resolve_takes_ctx(func) is False + + def test_keyword_only_param_not_detected(self) -> None: + """Keyword-only param with SkillContext annotation is not detected.""" + + def func(*, ctx: SkillContext) -> str: # type: ignore[type-arg] + return "" + + assert _resolve_takes_ctx(func) is False + + +class TestSkillsProviderCodeSkillBeforeRun: + """Tests for SkillsProvider before_run and combined scenarios.""" async def test_before_run_injects_code_skills(self) -> None: skill = Skill(name="prog-skill", description="A code-defined skill.", content="Body") diff --git a/python/samples/02-agents/skills/code_skill/README.md b/python/samples/02-agents/skills/code_skill/README.md index 4900d00eb5..d892400671 100644 --- a/python/samples/02-agents/skills/code_skill/README.md +++ b/python/samples/02-agents/skills/code_skill/README.md @@ -8,7 +8,7 @@ While file-based skills use `SKILL.md` files discovered on disk, code-defined sk 1. **Basic Code Skill** — Create a `Skill` directly with static resources (inline content) 2. **Dynamic Resources** — Attach callable resources via the `@skill.resource` decorator that generate content at invocation time -3. **Dynamic Resources with kwargs** — Attach a callable resource that accepts `**kwargs` to receive runtime arguments passed via `agent.run()`, useful for injecting request-scoped context (user tokens, session data) +3. **Typed Dependencies via `SkillContext`** — Declare a `SkillContext[DepsT]` first parameter on a resource function to receive typed dependencies injected by the `SkillsProvider` All patterns can be combined with file-based skills in a single `SkillsProvider`. @@ -48,7 +48,7 @@ uv run samples/02-agents/skills/code_skill/code_skill.py The sample runs two examples: 1. **Code style question** — Uses Pattern 1 (static resources): the agent loads the `code-style` skill and reads the `style-guide` resource to answer naming convention questions -2. **Project info question** — Uses Patterns 2 & 3 (dynamic resources with kwargs): the agent reads the dynamically generated `team-roster` resource and the `environment` resource which receives `app_version` via runtime kwargs +2. **Project info question** — Uses Patterns 2 & 3 (dynamic resources with `SkillContext`): the agent reads the dynamically generated `team-roster` resource and the `environment` resource which receives `app_version` via typed dependencies ## Learn More diff --git a/python/samples/02-agents/skills/code_skill/code_skill.py b/python/samples/02-agents/skills/code_skill/code_skill.py index e111567244..4f28e82c53 100644 --- a/python/samples/02-agents/skills/code_skill/code_skill.py +++ b/python/samples/02-agents/skills/code_skill/code_skill.py @@ -3,10 +3,11 @@ import asyncio import os import sys +from dataclasses import dataclass from textwrap import dedent from typing import Any -from agent_framework import Agent, Skill, SkillResource, SkillsProvider +from agent_framework import Agent, Skill, SkillContext, SkillResource, SkillsProvider from agent_framework.azure import AzureOpenAIResponsesClient from azure.identity import AzureCliCredential from dotenv import load_dotenv @@ -25,12 +26,17 @@ decorator. Resources can be sync or async functions that generate content at invocation time. -Pattern 3: Dynamic Resources with kwargs - Attach a callable resource that accepts **kwargs to receive runtime - arguments passed via agent.run(). This is useful for injecting - request-scoped context (user tokens, session data) into skill resources. +Pattern 3: Typed Dependencies via SkillContext + Pass deps to Skill so Pyright/mypy verify that @skill.resource callables + declare a matching SkillContext[DepsT] parameter. The provider injects a typed + context object at invocation time, giving the resource access to shared + dependencies (database clients, config, etc.). -Both patterns can be combined with file-based skills in a single SkillsProvider. +Pattern 4: Runtime kwargs + Resource functions with **kwargs receive runtime arguments forwarded from + agent.run(). Combine with SkillContext for both typed deps and dynamic args. + +All patterns can be combined with file-based skills in a single SkillsProvider. """ # Load environment variables from .env file @@ -67,26 +73,37 @@ ) # Pattern 2: Dynamic Resources — @skill.resource decorator +# Pattern 3: Typed Dependencies via SkillContext + + +@dataclass +class ProjectDeps: + """Shared dependencies for project-info skill resources.""" + + app_version: str = "2.4.1" + + +# By passing deps, Pyright/mypy verify that @skill.resource callables +# declare a matching SkillContext[ProjectDeps] parameter. project_info_skill = Skill( name="project-info", - description="Project status and configuration information", + description="Project status, configuration, team info, and request context", content=dedent("""\ Use this skill for questions about the current project status, - environment configuration, or team structure. + environment configuration, team structure, or request context. """), + deps=ProjectDeps(), ) @project_info_skill.resource -def environment(**kwargs: Any) -> str: +def environment(ctx: SkillContext[ProjectDeps]) -> str: """Get current environment configuration.""" - # Access runtime kwargs passed via agent.run(app_version="...") - app_version = kwargs.get("app_version", "unknown") env = os.environ.get("APP_ENV", "development") region = os.environ.get("APP_REGION", "us-east-1") return f"""\ # Environment Configuration - - App Version: {app_version} + - App Version: {ctx.deps.app_version} - Environment: {env} - Region: {region} - Python: {sys.version} @@ -106,6 +123,32 @@ def get_team_roster() -> str: """ +# Pattern 4: Runtime kwargs — resource receives arguments from agent.run() +@project_info_skill.resource(name="caller-info", description="Information about the caller") +def get_caller_info(ctx: SkillContext[ProjectDeps], **kwargs: Any) -> str: + """Return caller info combining typed deps and runtime kwargs.""" + caller = kwargs.get("caller_name", "unknown") + role = kwargs.get("caller_role", "unknown") + return f"""\ + # Caller Info + - Name: {caller} + - Role: {role} + - App Version: {ctx.deps.app_version} + """ + + +@project_info_skill.resource(name="request-context", description="Request context from kwargs only") +def get_request_context(**kwargs: Any) -> str: + """Return request context from runtime kwargs only (no SkillContext).""" + request_id = kwargs.get("request_id", "none") + client_agent = kwargs.get("client_agent", "unknown") + return f"""\ + # Request Context + - Request ID: {request_id} + - Client Agent: {client_agent} + """ + + async def main() -> None: """Run the code-defined skills demo.""" endpoint = os.environ["AZURE_AI_PROJECT_ENDPOINT"] @@ -133,11 +176,30 @@ async def main() -> None: response = await agent.run("What naming convention should I use for class attributes?") print(f"Agent: {response}\n") - # Example 2: Project info question (Pattern 2 & 3 — dynamic resources with kwargs) + # Example 2: Project info question (Patterns 2 & 3 — dynamic resources with SkillContext) print("Example 2: Project info question") print("---------------------------------") - # Pass app_version as a runtime kwarg; it flows to the environment() resource via **kwargs - response = await agent.run("What environment are we running in and who is on the team?", app_version="2.4.1") + response = await agent.run("What environment are we running in and who is on the team?") + print(f"Agent: {response}\n") + + # Example 3: kwargs forwarding (Pattern 4 — runtime kwargs reach resource functions) + print("Example 3: Caller info via kwargs") + print("----------------------------------") + response = await agent.run( + "Who is calling and what app version are we on?", + caller_name="Alice Chen", + caller_role="Tech Lead", + ) + print(f"Agent: {response}\n") + + # Example 4: kwargs-only resource (no SkillContext, just **kwargs) + print("Example 4: Request context via kwargs only") + print("-------------------------------------------") + response = await agent.run( + "What is the current request context?", + request_id="req-42", + client_agent="cli", + ) print(f"Agent: {response}\n") """ @@ -154,6 +216,14 @@ async def main() -> None: Agent: We're running app version 2.4.1 in the development environment in us-east-1. The team consists of Alice Chen (Tech Lead), Bob Smith (Backend Engineer), and Carol Davis (Frontend Engineer). + + Example 3: Caller info via kwargs + ---------------------------------- + Agent: The caller is Alice Chen (Tech Lead), running app version 2.4.1. + + Example 4: Request context via kwargs only + ------------------------------------------- + Agent: The current request context is Request ID req-42 from client agent cli. """