Python: Add Python feature lifecycle decorators for released APIs (#4975)

* Add Python feature lifecycle decorators

Introduce reusable experimental and release-candidate decorators for released packages, migrate the Skills APIs to the new staged metadata and warning system, and add lifecycle guidance plus samples.

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

* Fix Python CI follow-ups

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

* Address PR review feedback

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

* Preserve protocol runtime checks

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

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Eduard van Valkenburg
2026-03-31 21:40:08 +02:00
committed by GitHub
Unverified
parent 9c9d81d8b6
commit 55b6e7a9f4
19 changed files with 1220 additions and 60 deletions
+238
View File
@@ -0,0 +1,238 @@
# Copyright (c) Microsoft. All rights reserved.
---
name: python-feature-lifecycle
description: >
Guidance for package and feature lifecycle in the Agent Framework Python
codebase, including stage meanings, feature-stage decorators, feature enums,
and how to move APIs from one stage to the next.
---
# Python Feature Lifecycle
## Two lifecycle levels
Agent Framework uses lifecycle at two different levels:
1. **Package lifecycle** — the maturity of the package as a whole
2. **Feature lifecycle** — the maturity of a specific API or feature inside that package
These are related, but they are **not the same thing**.
- The **package stage is the default** for everything in the package.
- **Feature-stage decorators are only for exceptions** when a feature is behind the package's default stage.
- Do **not** decorate every class or function just because the package is experimental or release candidate.
### Important default
If a package is still in **beta / experimental preview**, all public APIs in that package are experimental by default.
- Do **not** add `@experimental(...)` everywhere in that package.
- The package stage already communicates that default.
Once a package moves forward, you can keep individual features behind:
- If a package moves to **release candidate**, a feature may remain **experimental**
- If a package moves to **released / GA**, a feature may remain **experimental** or **release candidate**
That is the main use case for feature-stage decorators.
## The four stages
### 1. Experimental
Use for features that are still unstable and may change or be removed without notice.
Feature-level code pattern:
```python
from ._feature_stage import ExperimentalFeature, experimental
@experimental(feature_id=ExperimentalFeature.MY_FEATURE)
class MyFeature:
...
```
Behavior:
- Adds an experimental warning block to the docstring
- Records feature metadata on the decorated object
- Emits a runtime warning the first time the feature is used (once per feature by default)
Enum setup:
- Add an all-caps member to `ExperimentalFeature`
- Reuse the same feature ID across all APIs that belong to the same conceptual feature
### 2. Release candidate
Use for features that are nearly stable but may still receive small refinements before GA.
Feature-level code pattern:
```python
from ._feature_stage import ReleaseCandidateFeature, release_candidate
@release_candidate(feature_id=ReleaseCandidateFeature.MY_FEATURE)
class MyFeature:
...
```
Behavior:
- Adds a release-candidate note to the docstring
- Records feature metadata on the decorated object
- Does **not** emit the experimental warning
Enum setup:
- Add an all-caps member to `ReleaseCandidateFeature`
### 3. Released
Use for stable GA APIs.
Code pattern:
- **No feature-stage decorator**
- **No entry** in `ExperimentalFeature`
- **No entry** in `ReleaseCandidateFeature`
If a feature is fully released, remove any stage-specific feature annotation.
### 4. Deprecated
Use for APIs that still exist but should not be used for new code.
Code pattern:
```python
import sys
if sys.version_info >= (3, 13):
from warnings import deprecated # type: ignore # pragma: no cover
else:
from typing_extensions import deprecated # type: ignore # pragma: no cover
@deprecated("MyOldFeature is deprecated. Use MyNewFeature instead.")
class MyOldFeature:
...
```
Behavior:
- Uses the repository's version-conditional deprecation import pattern
- Should describe what to use instead
Deprecated APIs should not also carry feature-stage decorators.
## Expected decorators by stage
| Feature stage | Expected annotation |
| --- | --- |
| Experimental | `@experimental(feature_id=ExperimentalFeature.X)` |
| Release candidate | `@release_candidate(feature_id=ReleaseCandidateFeature.X)` |
| Released | No feature-stage decorator |
| Deprecated | `@deprecated("...")` |
## Feature enums
The feature enums are the inventory of currently staged features:
- `ExperimentalFeature`
- `ReleaseCandidateFeature`
Guidance:
- Use one enum member per conceptual feature, not per class
- Ideally, an ADR already defines the overall feature boundary and therefore the feature ID that staged APIs for that feature should reuse
- Keep feature IDs all caps
- Reuse the same member across related APIs for the same feature
- Remove enum members when the feature no longer belongs to that stage
- Treat these enums as **current-stage inventories**, not as a stable consumer introspection API
Minimal consumer guidance:
- Treat `__feature_stage__` and `__feature_id__` as optional staged metadata, not as stable contracts
- Use `getattr(obj, "__feature_stage__", None)` and `getattr(obj, "__feature_id__", None)` rather than direct attribute access
- Treat missing metadata as "no explicit feature-stage annotation"
- For warning filters while a feature is staged, match the literal feature ID string
- Do **not** rely on `ExperimentalFeature.X`, `ReleaseCandidateFeature.X`, or the continued presence of `__feature_id__` after a feature moves stages or is released
For consumers, the enums are also re-exported from `agent_framework`.
For internal implementation code inside `agent_framework`, continue to import the enums and decorators from `._feature_stage`.
## Package stage vs feature stage
Use the following rules:
### Package is experimental / beta
- All public APIs are experimental by default
- Do **not** add feature-stage decorators just to restate that
- Only introduce feature-level annotations later if the package advances first
### Package is release candidate
- All public APIs are RC by default
- Do **not** decorate everything
- Add `@experimental(...)` only for features that are intentionally still behind the package
### Package is released / GA
- All public APIs are released by default
- Add `@experimental(...)` or `@release_candidate(...)` only for features still being held back
## Moving a feature from one stage to the next
### Experimental -> Release candidate
1. Move the feature ID from `ExperimentalFeature` to `ReleaseCandidateFeature`
2. Replace `@experimental(...)` with `@release_candidate(...)`
3. Update any tests or docs that mention the old stage
### Experimental -> Released
1. Remove `@experimental(...)`
2. Remove the feature from `ExperimentalFeature`
3. Do not add a replacement feature-stage decorator
### Release candidate -> Released
1. Remove `@release_candidate(...)`
2. Remove the feature from `ReleaseCandidateFeature`
3. Leave the API undecorated
### Any stage -> Deprecated
1. Remove any feature-stage decorator
2. Remove the feature from the stage enum
3. Add `@deprecated("...")`
4. Update docs/tests to reflect the replacement path
## Promotion guidance
Features do **not** have to pass through every stage.
- It is usually a good idea to move features in order when that reflects reality
- But it is completely acceptable to go **experimental -> released**
- Do **not** force a feature through release candidate if there is no real RC period
Likewise, when a package advances, do not automatically move every feature with it.
- Promote features based on actual readiness
- Keep lagging features explicitly marked only when they are behind the package default
## Practical rules of thumb
- **Package default first, feature exceptions second**
- **Do not decorate everything in preview packages**
- **Do not double-annotate members of an already-staged class**
- **Use enums only for currently staged features**
- **Do not treat stage enums as a compatibility contract**
- **Treat `__feature_stage__` and `__feature_id__` as optional metadata; use `getattr`**
- **Remove stage annotations once a feature is released or deprecated**
+1
View File
@@ -11,6 +11,7 @@ Instructions for AI coding agents working in the Python codebase.
- `python-development` — coding standards, type annotations, docstrings, logging, performance
- `python-testing` — test structure, fixtures, async mode, running tests
- `python-code-quality` — linting, formatting, type checking, prek hooks, CI workflow
- `python-feature-lifecycle` — package vs feature lifecycle stages, decorators, enums, and promotion guidance
- `python-package-management` — monorepo structure, lazy loading, versioning, new packages
- `python-samples` — sample file structure, PEP 723, documentation guidelines
@@ -76,8 +76,8 @@ def mock_project_client() -> MagicMock:
mock_client.telemetry = MagicMock()
mock_client.telemetry.get_application_insights_connection_string = AsyncMock()
# Mock get_openai_client method
mock_client.get_openai_client = MagicMock()
# AIProjectClient.get_openai_client() is a sync accessor, even on the aio client.
mock_client.get_openai_client = MagicMock(return_value=MagicMock())
# Mock close method
mock_client.close = AsyncMock()
@@ -33,8 +33,8 @@ def mock_project_client() -> MagicMock:
mock_client.telemetry = MagicMock()
mock_client.telemetry.get_application_insights_connection_string = AsyncMock()
# Mock get_openai_client method
mock_client.get_openai_client = MagicMock()
# AIProjectClient.get_openai_client() is a sync accessor, even on the aio client.
mock_client.get_openai_client = MagicMock(return_value=MagicMock())
# Mock close method
mock_client.close = AsyncMock()
@@ -78,6 +78,7 @@ from ._evaluation import (
tool_called_check,
tool_calls_present,
)
from ._feature_stage import ExperimentalFeature, ReleaseCandidateFeature
from ._mcp import MCPStdioTool, MCPStreamableHTTPTool, MCPWebsocketTool
from ._middleware import (
AgentContext,
@@ -314,6 +315,7 @@ __all__ = [
"Evaluator",
"Executor",
"ExpectedToolCall",
"ExperimentalFeature",
"FanInEdgeGroup",
"FanOutEdgeGroup",
"FileCheckpointStorage",
@@ -344,6 +346,7 @@ __all__ = [
"OuterFinalT",
"OuterUpdateT",
"RawAgent",
"ReleaseCandidateFeature",
"ResponseStream",
"Role",
"RoleLiteral",
@@ -3,12 +3,14 @@
from __future__ import annotations
import inspect
import textwrap
from collections.abc import Callable, Mapping
from typing import Any
_GOOGLE_SECTION_HEADERS = (
"Args:",
"Keyword Args:",
"Attributes:",
"Returns:",
"Raises:",
"Examples:",
@@ -45,6 +47,29 @@ def _format_keyword_arg_lines(extra_keyword_args: Mapping[str, str]) -> list[str
return formatted_lines
def insert_docstring_block(docstring: str | None, *, block: str) -> str | None:
"""Insert a preformatted block before the first Google-style section."""
cleaned_block = textwrap.dedent(block).strip()
if not cleaned_block:
return docstring
if not docstring:
return cleaned_block
lines = inspect.cleandoc(docstring).splitlines()
block_lines = cleaned_block.splitlines()
insert_index = _find_next_section_index(lines, 0)
insertion: list[str] = []
if insert_index > 0 and lines[insert_index - 1] != "":
insertion.append("")
insertion.extend(block_lines)
if insert_index < len(lines) and insertion[-1] != "":
insertion.append("")
lines[insert_index:insert_index] = insertion
return "\n".join(lines).rstrip()
def build_layered_docstring(
source: Callable[..., Any],
*,
@@ -0,0 +1,278 @@
# Copyright (c) Microsoft. All rights reserved.
from __future__ import annotations
import asyncio.coroutines
import functools
import inspect
import sys
import warnings
from collections.abc import Callable
from enum import Enum
from types import MethodType
from typing import Any, Literal, TypeVar, cast
from ._docstrings import insert_docstring_block
FeatureStageT = TypeVar("FeatureStageT", bound=Callable[..., Any])
FeatureStageName = Literal["experimental", "release_candidate"]
# Optional feature-stage metadata for warnings and best-effort introspection.
_FEATURE_ID_ATTR = "__feature_id__"
_FEATURE_STAGE_ATTR = "__feature_stage__"
_WARNED_FEATURES: set[tuple[type[Warning], str]] = set()
_EXPERIMENTAL_DOCSTRING = """\
.. warning:: Experimental
This API is experimental and subject to change or removal
in future versions without notice.
"""
_RELEASE_CANDIDATE_DOCSTRING = """\
.. note:: Release candidate
This API is in release-candidate stage and may receive
minor refinements before it is considered generally available.
"""
class ExperimentalFeature(str, Enum):
"""Current experimental feature IDs.
This enum is a stage-scoped inventory, not a stable introspection surface.
Members may move or be removed as features advance. The `__feature_id__`
attribute is also optional stage metadata and may disappear when a feature
is released, so consumer code should use `getattr(...)` rather than relying
on enum membership or attribute presence over time.
"""
SKILLS = "SKILLS"
class ReleaseCandidateFeature(str, Enum):
"""Current release-candidate feature IDs.
This enum is a stage-scoped inventory, not a stable introspection surface.
Members may move or be removed as features advance. The `__feature_id__`
attribute is also optional stage metadata and may disappear when a feature
is released, so consumer code should use `getattr(...)` rather than relying
on enum membership or attribute presence over time.
"""
class FeatureStageWarning(FutureWarning):
"""Base warning category for staged APIs."""
class ExperimentalWarning(FeatureStageWarning):
"""Warning emitted when an experimental API is used."""
def _normalize_feature_id(feature_id: str | Enum) -> str:
return str(feature_id.value if isinstance(feature_id, Enum) else feature_id)
def _get_object_name(obj: Any) -> str:
return str(getattr(obj, "__qualname__", getattr(obj, "__name__", type(obj).__name__)))
def _get_descriptor_callable(obj: Any) -> Callable[..., Any]:
return cast(Callable[..., Any], obj.__func__)
def _is_protocol_class(obj: Any) -> bool:
return isinstance(obj, type) and bool(getattr(obj, "_is_protocol", False))
def _build_stage_warning_message(*, stage: FeatureStageName, feature_id: str, object_name: str) -> str:
if stage == "experimental":
return (
f"[{feature_id}] {object_name} is experimental and may change or be removed in future versions "
"without notice."
)
return (
f"[{feature_id}] {object_name} is in release-candidate stage and may receive minor refinements before it is "
"considered generally available."
)
def _set_feature_stage_metadata(obj: Any, *, stage: FeatureStageName, feature_id: str) -> None:
setattr(obj, _FEATURE_STAGE_ATTR, stage)
setattr(obj, _FEATURE_ID_ATTR, feature_id)
def _warn_on_feature_use(
*,
stage: FeatureStageName,
feature_id: str,
object_name: str,
category: type[Warning],
stacklevel: int,
) -> None:
warning_key = (category, feature_id)
if warning_key in _WARNED_FEATURES:
return
warnings.warn(
_build_stage_warning_message(stage=stage, feature_id=feature_id, object_name=object_name),
category=category,
stacklevel=stacklevel,
)
_WARNED_FEATURES.add(warning_key)
def _add_runtime_warning(
obj: FeatureStageT,
*,
stage: FeatureStageName,
feature_id: str,
category: type[Warning],
) -> FeatureStageT:
object_name = _get_object_name(obj)
if isinstance(obj, type):
experimental_class = cast(type[Any], obj)
original_new: Any = experimental_class.__new__
@functools.wraps(original_new)
def __new__(cls: type[Any], /, *args: Any, **kwargs: Any) -> Any:
if cls is experimental_class:
_warn_on_feature_use(
stage=stage,
feature_id=feature_id,
object_name=object_name,
category=category,
stacklevel=3,
)
if original_new is not object.__new__:
return original_new(cls, *args, **kwargs)
if cls.__init__ is object.__init__ and (args or kwargs):
raise TypeError(f"{cls.__name__}() takes no arguments")
return original_new(cls)
experimental_class.__new__ = staticmethod(__new__) # type: ignore[assignment]
original_init_subclass: Any = experimental_class.__init_subclass__
if isinstance(original_init_subclass, MethodType):
original_init_subclass_func = original_init_subclass.__func__
@functools.wraps(original_init_subclass_func)
def bound_init_subclass_wrapper(*args: Any, **kwargs: Any) -> Any:
_warn_on_feature_use(
stage=stage,
feature_id=feature_id,
object_name=object_name,
category=category,
stacklevel=3,
)
return original_init_subclass_func(*args, **kwargs)
experimental_class.__init_subclass__ = classmethod(bound_init_subclass_wrapper) # type: ignore[assignment]
else:
@functools.wraps(original_init_subclass)
def init_subclass_wrapper(*args: Any, **kwargs: Any) -> Any:
_warn_on_feature_use(
stage=stage,
feature_id=feature_id,
object_name=object_name,
category=category,
stacklevel=3,
)
return original_init_subclass(*args, **kwargs)
experimental_class.__init_subclass__ = init_subclass_wrapper # type: ignore[assignment]
return cast(FeatureStageT, experimental_class)
@functools.wraps(obj)
def wrapper(*args: Any, **kwargs: Any) -> Any:
_warn_on_feature_use(
stage=stage,
feature_id=feature_id,
object_name=object_name,
category=category,
stacklevel=3,
)
return obj(*args, **kwargs)
if inspect.iscoroutinefunction(obj):
if sys.version_info >= (3, 12):
wrapper = inspect.markcoroutinefunction(wrapper)
else:
wrapper._is_coroutine = asyncio.coroutines._is_coroutine # type: ignore[attr-defined]
return cast(FeatureStageT, wrapper)
def _feature_stage(
*,
stage: FeatureStageName,
feature_id: str | Enum,
docstring_block: str,
warning_category: type[Warning] | None,
) -> Callable[[FeatureStageT], FeatureStageT]:
normalized_feature_id = _normalize_feature_id(feature_id)
def decorator(obj: FeatureStageT) -> FeatureStageT:
descriptor_wrapper: Callable[[Any], Any] | None = None
target: Any = obj
if isinstance(obj, staticmethod):
descriptor_wrapper = staticmethod
target = _get_descriptor_callable(obj)
elif isinstance(obj, classmethod):
descriptor_wrapper = classmethod
target = _get_descriptor_callable(obj)
if not callable(target):
raise TypeError(f"{stage} decorator can only be applied to classes and callables, not {obj!r}.")
is_protocol_class = _is_protocol_class(target)
decorated: Any = target
if warning_category is not None and not is_protocol_class:
decorated = _add_runtime_warning(
target,
stage=stage,
feature_id=normalized_feature_id,
category=warning_category,
)
updated_docstring = insert_docstring_block(decorated.__doc__, block=docstring_block)
if updated_docstring is not None:
decorated.__doc__ = updated_docstring
# runtime_checkable Protocol classes treat added class attributes as protocol members
# on older Python versions, which breaks isinstance/issubclass checks.
if not is_protocol_class:
_set_feature_stage_metadata(decorated, stage=stage, feature_id=normalized_feature_id)
if descriptor_wrapper is not None:
return cast(FeatureStageT, descriptor_wrapper(decorated))
return cast(FeatureStageT, decorated)
return decorator
def experimental(*, feature_id: ExperimentalFeature) -> Callable[[FeatureStageT], FeatureStageT]:
"""Mark a class or callable as experimental."""
return _feature_stage(
stage="experimental",
feature_id=feature_id,
docstring_block=_EXPERIMENTAL_DOCSTRING,
warning_category=ExperimentalWarning,
)
def release_candidate(
*,
feature_id: ReleaseCandidateFeature,
) -> Callable[[FeatureStageT], FeatureStageT]:
"""Mark a class or callable as release-candidate."""
return _feature_stage(
stage="release_candidate",
feature_id=feature_id,
docstring_block=_RELEASE_CANDIDATE_DOCSTRING,
warning_category=None,
)
@@ -35,6 +35,7 @@ from html import escape as xml_escape
from pathlib import Path, PurePosixPath
from typing import TYPE_CHECKING, Any, ClassVar, Final, Protocol, runtime_checkable
from ._feature_stage import ExperimentalFeature, experimental
from ._sessions import BaseContextProvider
from ._tools import FunctionTool
@@ -47,14 +48,10 @@ logger = logging.getLogger(__name__)
# region Models
@experimental(feature_id=ExperimentalFeature.SKILLS)
class SkillResource:
"""A named piece of supplementary content attached to a skill.
.. warning:: Experimental
This API is experimental and subject to change or removal
in future versions without notice.
A resource provides data that an agent can retrieve on demand. It holds
either a static ``content`` string or a ``function`` that produces content
dynamically (sync or async). Exactly one must be provided.
@@ -117,14 +114,10 @@ class SkillResource:
self._accepts_kwargs = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values())
@experimental(feature_id=ExperimentalFeature.SKILLS)
class SkillScript:
"""An executable script attached to a skill.
.. warning:: Experimental
This API is experimental and subject to change or removal
in future versions without notice.
A script represents executable code that an agent can run. It holds
either an inline ``function`` callable (code-defined scripts) or
a ``path`` to a script file on disk (file-based scripts).
@@ -202,11 +195,6 @@ class SkillScript:
def parameters_schema(self) -> dict[str, Any] | None:
"""JSON Schema describing the script's parameters.
.. warning:: Experimental
This API is experimental and subject to change or removal
in future versions without notice.
Lazily generated from the callable's signature on first access.
Returns ``None`` for file-based scripts or functions with no
introspectable parameters.
@@ -219,14 +207,10 @@ class SkillScript:
return self._parameters_schema
@experimental(feature_id=ExperimentalFeature.SKILLS)
class Skill:
"""A skill definition with optional resources.
.. warning:: Experimental
This API is experimental and subject to change or removal
in future versions without notice.
A skill bundles a set of instructions (``content``) with metadata and
zero or more :class:`SkillResource` and :class:`SkillScript` instances.
Resources and scripts can be supplied at construction time or added later
@@ -432,14 +416,10 @@ class Skill:
@runtime_checkable
@experimental(feature_id=ExperimentalFeature.SKILLS)
class SkillScriptRunner(Protocol):
"""Protocol for skill script runners.
.. warning:: Experimental
This API is experimental and subject to change or removal
in future versions without notice.
A script runner determines how **file-based** skill scripts are
run. Implementations decide the execution strategy
(e.g., local subprocess, hosted code execution environment,
@@ -538,14 +518,10 @@ SCRIPT_RUNNER_INSTRUCTIONS: Final[str] = (
# region SkillsProvider
@experimental(feature_id=ExperimentalFeature.SKILLS)
class SkillsProvider(BaseContextProvider):
"""Context provider that advertises skills and exposes skill tools.
.. warning:: Experimental
This API is experimental and subject to change or removal
in future versions without notice.
Supports both **file-based** skills (discovered from ``SKILL.md`` files)
and **code-defined** skills (passed as :class:`Skill` instances).
+10 -3
View File
@@ -3,6 +3,7 @@
import asyncio
import logging
import sys
import warnings
from collections.abc import AsyncIterable, Awaitable, MutableSequence, Sequence
from typing import Any, Generic
from unittest.mock import patch
@@ -10,7 +11,13 @@ from uuid import uuid4
from pytest import fixture
from agent_framework import (
warnings.filterwarnings(
"ignore",
message=r"\[SKILLS\].*",
category=FutureWarning,
)
from agent_framework import ( # noqa: E402
AgentResponse,
AgentResponseUpdate,
AgentSession,
@@ -26,8 +33,8 @@ from agent_framework import (
SupportsAgentRun,
tool,
)
from agent_framework._clients import OptionsCoT
from agent_framework.observability import ChatTelemetryLayer
from agent_framework._clients import OptionsCoT # noqa: E402
from agent_framework.observability import ChatTelemetryLayer # noqa: E402
if sys.version_info >= (3, 12):
from typing import override # type: ignore
@@ -1,6 +1,6 @@
# Copyright (c) Microsoft. All rights reserved.
from agent_framework._docstrings import apply_layered_docstring, build_layered_docstring
from agent_framework._docstrings import apply_layered_docstring, build_layered_docstring, insert_docstring_block
# -- Helpers: stub functions with various docstring shapes --
@@ -36,6 +36,14 @@ def _source_no_sections() -> None:
"""A plain summary with no Google-style sections."""
def _source_with_attributes() -> None:
"""A documented object.
Attributes:
value: A documented attribute.
"""
def _source_no_docstring() -> None:
pass
@@ -141,6 +149,67 @@ def test_build_preserves_multiple_extra_kwargs_order() -> None:
assert alpha_idx < beta_idx < gamma_idx
# -- insert_docstring_block tests --
def test_insert_docstring_block_before_args_section() -> None:
result = insert_docstring_block(
_source_with_args_only.__doc__,
block="""\
.. warning:: Experimental
This API is experimental.
""",
)
assert result is not None
lines = result.splitlines()
warning_index = next(i for i, line in enumerate(lines) if line == ".. warning:: Experimental")
args_index = next(i for i, line in enumerate(lines) if line == "Args:")
assert warning_index < args_index
def test_insert_docstring_block_before_attributes_section() -> None:
result = insert_docstring_block(
_source_with_attributes.__doc__,
block="""\
.. warning:: Experimental
This API is experimental.
""",
)
assert result is not None
lines = result.splitlines()
warning_index = next(i for i, line in enumerate(lines) if line == ".. warning:: Experimental")
attributes_index = next(i for i, line in enumerate(lines) if line == "Attributes:")
assert warning_index < attributes_index
def test_insert_docstring_block_appends_when_no_sections() -> None:
result = insert_docstring_block(
_source_no_sections.__doc__,
block="""\
.. note:: Release candidate
This API is nearly final.
""",
)
assert result is not None
assert result.endswith("This API is nearly final.")
assert ".. note:: Release candidate" in result
def test_insert_docstring_block_returns_block_for_missing_docstring() -> None:
result = insert_docstring_block(
_source_no_docstring.__doc__,
block="""\
.. warning:: Experimental
This API is experimental.
""",
)
assert result == ".. warning:: Experimental\n\n This API is experimental."
# -- apply_layered_docstring tests --
@@ -0,0 +1,427 @@
# Copyright (c) Microsoft. All rights reserved.
from __future__ import annotations
import inspect
import warnings
from enum import Enum
from typing import Protocol, runtime_checkable
import pytest
from agent_framework import ExperimentalFeature as PublicExperimentalFeature
from agent_framework import ReleaseCandidateFeature as PublicReleaseCandidateFeature
from agent_framework._feature_stage import (
_WARNED_FEATURES,
ExperimentalWarning,
_feature_stage,
experimental,
release_candidate,
)
from agent_framework._feature_stage import (
ExperimentalFeature as InternalExperimentalFeature,
)
from agent_framework._feature_stage import (
ReleaseCandidateFeature as InternalReleaseCandidateFeature,
)
class AlternateExperimentalFeature(str, Enum):
EXPERIMENTAL_FEATURE = "EXPERIMENTAL_FEATURE"
SHARED_FEATURE = "SHARED_EXPERIMENTAL_FEATURE"
ALTERNATE_FEATURE = "ALTERNATE_EXPERIMENTAL_FEATURE"
class InvalidStageFeature(str, Enum):
LOWERCASE = "skills"
class NonStringFeature(Enum):
INTEGER = 1
class HelperReleaseCandidateFeature(str, Enum):
RC_FEATURE = "RC_FEATURE"
@pytest.fixture(autouse=True)
def clear_feature_warning_state() -> None:
_WARNED_FEATURES.clear()
yield
_WARNED_FEATURES.clear()
def test_feature_enums_are_exposed_from_root() -> None:
assert PublicExperimentalFeature is InternalExperimentalFeature
assert PublicReleaseCandidateFeature is InternalReleaseCandidateFeature
def test_experimental_decorator_accepts_feature_enum() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@experimental(feature_id=AlternateExperimentalFeature.EXPERIMENTAL_FEATURE) # type: ignore[arg-type]
def skill_function() -> None:
pass
assert not caught
with warnings.catch_warnings(record=True) as caught:
skill_function()
assert len(caught) == 1
assert f"[{AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value}]" in str(caught[0].message)
assert "skill_function" in str(caught[0].message)
assert skill_function.__feature_id__ == AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value
def test_experimental_function_warns_on_call_and_not_on_definition() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@experimental(feature_id=AlternateExperimentalFeature.EXPERIMENTAL_FEATURE) # type: ignore[arg-type]
def my_function(value: int) -> int:
"""Double the input.
Args:
value: Value to double.
Returns:
The doubled value.
"""
return value * 2
assert not caught
with warnings.catch_warnings(record=True) as caught:
assert my_function(3) == 6
assert my_function(4) == 8
assert len(caught) == 1
assert f"[{AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value}]" in str(caught[0].message)
assert "my_function" in str(caught[0].message)
assert my_function.__feature_stage__ == "experimental"
assert my_function.__feature_id__ == AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value
assert my_function.__doc__ is not None
lines = my_function.__doc__.splitlines()
warning_index = next(i for i, line in enumerate(lines) if line == ".. warning:: Experimental")
args_index = next(i for i, line in enumerate(lines) if line == "Args:")
assert warning_index < args_index
def test_experimental_class_warns_on_instantiation_and_not_on_definition() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@experimental(feature_id=AlternateExperimentalFeature.EXPERIMENTAL_FEATURE) # type: ignore[arg-type]
class ExperimentalClass:
"""An experimental class.
Args:
value: Value to store.
"""
def __init__(self, value: int) -> None:
self.value = value
assert not caught
with warnings.catch_warnings(record=True) as caught:
instantiation_line = inspect.currentframe().f_lineno + 1
instance = ExperimentalClass(4)
second_instance = ExperimentalClass(5)
assert len(caught) == 1
assert f"[{AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value}]" in str(caught[0].message)
assert "ExperimentalClass" in str(caught[0].message)
assert caught[0].filename == __file__
assert caught[0].lineno == instantiation_line
assert instance.value == 4
assert second_instance.value == 5
assert ExperimentalClass.__feature_stage__ == "experimental"
assert ExperimentalClass.__feature_id__ == AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value
def test_experimental_runtime_checkable_protocol_keeps_protocol_runtime_checks() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@runtime_checkable
@experimental(feature_id=AlternateExperimentalFeature.EXPERIMENTAL_FEATURE) # type: ignore[arg-type]
class ExampleProtocol(Protocol):
"""A protocol used for runtime checks.
Returns:
Nothing.
"""
def __call__(self, value: int) -> int: ...
assert not caught
def implementation(value: int) -> int:
return value
assert isinstance(implementation, ExampleProtocol)
assert ExampleProtocol.__doc__ is not None
assert ".. warning:: Experimental" in ExampleProtocol.__doc__
assert getattr(ExampleProtocol, "__feature_stage__", None) is None
assert getattr(ExampleProtocol, "__feature_id__", None) is None
def test_experimental_warning_is_emitted_once_per_feature() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@experimental(feature_id=AlternateExperimentalFeature.SHARED_FEATURE) # type: ignore[arg-type]
def first() -> None:
pass
@experimental(feature_id=AlternateExperimentalFeature.SHARED_FEATURE) # type: ignore[arg-type]
class Second:
pass
assert not caught
with warnings.catch_warnings(record=True) as caught:
first()
Second()
assert first is not None
assert Second is not None
assert len(caught) == 1
assert f"[{AlternateExperimentalFeature.SHARED_FEATURE.value}]" in str(caught[0].message)
assert "first" in str(caught[0].message)
def test_release_candidate_internal_helper_adds_metadata_without_runtime_warning() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@_feature_stage(
stage="release_candidate",
feature_id=HelperReleaseCandidateFeature.RC_FEATURE,
docstring_block="""\
.. note:: Release candidate
This API is in release-candidate stage and may receive
minor refinements before it is considered generally available.
""",
warning_category=None,
)
class ReleaseCandidateClass:
"""A release-candidate class.
Args:
value: Value to store.
"""
def __init__(self, value: int) -> None:
self.value = value
assert not caught
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
instance = ReleaseCandidateClass(5)
assert instance.value == 5
assert not caught
assert ReleaseCandidateClass.__feature_stage__ == "release_candidate"
assert ReleaseCandidateClass.__feature_id__ == HelperReleaseCandidateFeature.RC_FEATURE.value
assert ReleaseCandidateClass.__doc__ is not None
assert ".. note:: Release candidate" in ReleaseCandidateClass.__doc__
def test_experimental_property_warns_on_access_and_not_on_definition() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
class Example:
@property
@experimental(feature_id=AlternateExperimentalFeature.EXPERIMENTAL_FEATURE) # type: ignore[arg-type]
def value(self) -> int:
"""Return the value.
Returns:
The stored value.
"""
return 1
assert not caught
with warnings.catch_warnings(record=True) as caught:
assert Example().value == 1
assert Example().value == 1
assert len(caught) == 1
assert f"[{AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value}]" in str(caught[0].message)
assert "Example.value" in str(caught[0].message)
assert Example.value.__doc__ is not None
lines = Example.value.__doc__.splitlines()
warning_index = next(i for i, line in enumerate(lines) if line == ".. warning:: Experimental")
returns_index = next(i for i, line in enumerate(lines) if line == "Returns:")
assert warning_index < returns_index
def test_experimental_staticmethod_warns_when_decorator_wraps_descriptor() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
class Example:
@experimental(feature_id=AlternateExperimentalFeature.EXPERIMENTAL_FEATURE) # type: ignore[arg-type]
@staticmethod
def value() -> int:
"""Return the value.
Returns:
The stored value.
"""
return 1
assert not caught
with warnings.catch_warnings(record=True) as caught:
assert Example.value() == 1
assert Example.value() == 1
assert len(caught) == 1
assert f"[{AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value}]" in str(caught[0].message)
assert "Example.value" in str(caught[0].message)
assert Example.value.__feature_id__ == AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value
assert Example.value.__doc__ is not None
lines = Example.value.__doc__.splitlines()
warning_index = next(i for i, line in enumerate(lines) if line == ".. warning:: Experimental")
returns_index = next(i for i, line in enumerate(lines) if line == "Returns:")
assert warning_index < returns_index
def test_experimental_classmethod_warns_when_decorator_wraps_descriptor() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
class Example:
@experimental(feature_id=AlternateExperimentalFeature.EXPERIMENTAL_FEATURE) # type: ignore[arg-type]
@classmethod
def value(cls) -> int:
"""Return the value.
Returns:
The stored value.
"""
return 1
assert not caught
with warnings.catch_warnings(record=True) as caught:
assert Example.value() == 1
assert Example.value() == 1
assert len(caught) == 1
assert f"[{AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value}]" in str(caught[0].message)
assert "Example.value" in str(caught[0].message)
assert Example.value.__func__.__feature_id__ == AlternateExperimentalFeature.EXPERIMENTAL_FEATURE.value
assert Example.value.__doc__ is not None
lines = Example.value.__doc__.splitlines()
warning_index = next(i for i, line in enumerate(lines) if line == ".. warning:: Experimental")
returns_index = next(i for i, line in enumerate(lines) if line == "Returns:")
assert warning_index < returns_index
def test_feature_id_allows_lowercase_values() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@_feature_stage(
stage="experimental",
feature_id=InvalidStageFeature.LOWERCASE,
docstring_block=".. warning:: Experimental",
warning_category=ExperimentalWarning,
)
def lowercase_feature() -> None:
pass
assert not caught
with warnings.catch_warnings(record=True) as caught:
lowercase_feature()
assert len(caught) == 1
assert "[skills]" in str(caught[0].message)
assert "lowercase_feature" in str(caught[0].message)
assert lowercase_feature.__feature_id__ == "skills"
def test_experimental_decorator_allows_string_feature_id_at_runtime() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@experimental(feature_id="STRING_FEATURE") # type: ignore[arg-type]
def skill_function() -> None:
pass
assert not caught
with warnings.catch_warnings(record=True) as caught:
skill_function()
assert len(caught) == 1
assert "[STRING_FEATURE]" in str(caught[0].message)
assert "skill_function" in str(caught[0].message)
assert skill_function.__feature_id__ == "STRING_FEATURE"
def test_experimental_decorator_allows_other_enum_values_at_runtime() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@experimental(feature_id=AlternateExperimentalFeature.ALTERNATE_FEATURE) # type: ignore[arg-type]
def my_function() -> None:
pass
assert not caught
with warnings.catch_warnings(record=True) as caught:
my_function()
assert len(caught) == 1
assert f"[{AlternateExperimentalFeature.ALTERNATE_FEATURE.value}]" in str(caught[0].message)
assert "my_function" in str(caught[0].message)
assert my_function.__feature_id__ == AlternateExperimentalFeature.ALTERNATE_FEATURE.value
def test_release_candidate_decorator_allows_string_feature_id_at_runtime() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@release_candidate(feature_id="RC_FEATURE") # type: ignore[arg-type]
class ReleaseCandidateClass:
"""A release-candidate class."""
assert not caught
assert ReleaseCandidateClass.__feature_stage__ == "release_candidate"
assert ReleaseCandidateClass.__feature_id__ == "RC_FEATURE"
def test_feature_id_stringifies_non_string_enum_values() -> None:
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always")
@_feature_stage(
stage="experimental",
feature_id=NonStringFeature.INTEGER,
docstring_block=".. warning:: Experimental",
warning_category=ExperimentalWarning,
)
def numeric_feature() -> None:
pass
assert not caught
with warnings.catch_warnings(record=True) as caught:
numeric_feature()
assert len(caught) == 1
assert "[1]" in str(caught[0].message)
assert "numeric_feature" in str(caught[0].message)
assert numeric_feature.__feature_id__ == "1"
+41 -15
View File
@@ -11,7 +11,7 @@ from unittest.mock import AsyncMock
import pytest
from agent_framework import SessionContext, Skill, SkillResource, SkillsProvider
from agent_framework import SessionContext, Skill, SkillResource, SkillScript, SkillScriptRunner, SkillsProvider
from agent_framework._skills import (
DEFAULT_RESOURCE_EXTENSIONS,
DEFAULT_SCRIPT_EXTENSIONS,
@@ -32,6 +32,8 @@ from agent_framework._skills import (
_validate_skill_metadata,
)
pytestmark = pytest.mark.filterwarnings(r"ignore:\[SKILLS\].*:FutureWarning")
async def _noop_script_runner(skill: Any, script: Any, args: Any = None) -> None:
"""No-op script runner for tests that need a SkillScriptRunner."""
@@ -778,6 +780,44 @@ class TestSymlinkDetection:
# ---------------------------------------------------------------------------
class TestSkillsExperimentalStage:
"""Tests for the experimental stage annotations applied to skills APIs."""
def test_docstrings_include_experimental_warning(self) -> None:
assert SkillResource.__doc__ is not None
assert SkillScript.__doc__ is not None
assert Skill.__doc__ is not None
assert SkillScriptRunner.__doc__ is not None
assert SkillsProvider.__doc__ is not None
assert SkillScript.parameters_schema.__doc__ is not None
assert ".. warning:: Experimental" in SkillResource.__doc__
assert ".. warning:: Experimental" in SkillScript.__doc__
assert ".. warning:: Experimental" in Skill.__doc__
assert ".. warning:: Experimental" in SkillScriptRunner.__doc__
assert ".. warning:: Experimental" in SkillsProvider.__doc__
assert ".. warning:: Experimental" not in SkillScript.parameters_schema.__doc__
def test_feature_metadata_is_set(self) -> None:
assert SkillResource.__feature_stage__ == "experimental"
assert SkillScript.__feature_stage__ == "experimental"
assert Skill.__feature_stage__ == "experimental"
assert SkillsProvider.__feature_stage__ == "experimental"
feature_ids = [
SkillResource.__feature_id__,
SkillScript.__feature_id__,
Skill.__feature_id__,
SkillsProvider.__feature_id__,
]
assert all(isinstance(feature_id, str) and feature_id for feature_id in feature_ids)
assert len(set(feature_ids)) == 1
assert getattr(SkillScriptRunner, "__feature_stage__", None) is None
assert getattr(SkillScriptRunner, "__feature_id__", None) is None
assert SkillScript.parameters_schema.fget is not None
assert not hasattr(SkillScript.parameters_schema.fget, "__feature_stage__")
assert not hasattr(SkillScript.parameters_schema.fget, "__feature_id__")
class TestSkillResource:
"""Tests for SkillResource dataclass."""
@@ -1839,40 +1879,28 @@ class TestSkillScript:
"""Tests for the SkillScript data model."""
def test_empty_name_raises(self) -> None:
from agent_framework import SkillScript
with pytest.raises(ValueError, match="Script name cannot be empty"):
SkillScript(name="")
def test_whitespace_name_raises(self) -> None:
from agent_framework import SkillScript
with pytest.raises(ValueError, match="Script name cannot be empty"):
SkillScript(name=" ")
def test_path_default_none(self) -> None:
from agent_framework import SkillScript
script = SkillScript(name="test", function=lambda: None)
assert script.path is None
def test_path_set_explicitly(self) -> None:
from agent_framework import SkillScript
script = SkillScript(name="gen.py", path="/skills/my-skill/scripts/gen.py")
assert script.path == "/skills/my-skill/scripts/gen.py"
def test_create_with_function(self) -> None:
from agent_framework import SkillScript
script = SkillScript(name="analyze", description="Run analysis", function=lambda: "result")
assert script.name == "analyze"
assert script.description == "Run analysis"
assert script.function is not None
def test_accepts_kwargs_true_for_kwargs_function(self) -> None:
from agent_framework import SkillScript
def func_with_kwargs(**kwargs: Any) -> str:
return "result"
@@ -1880,8 +1908,6 @@ class TestSkillScript:
assert script._accepts_kwargs is True
def test_accepts_kwargs_false_for_regular_function(self) -> None:
from agent_framework import SkillScript
def func_no_kwargs(x: int = 0) -> str:
return "result"
@@ -0,0 +1,83 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from typing import Any
import agent_framework
"""Feature stage introspection.
This sample demonstrates how to inspect feature lifecycle metadata on Agent
Framework APIs.
The recommended minimal setup for consumers is:
1. Read `__feature_stage__` with `getattr(...)` to see whether an API is staged
2. Read `__feature_id__` with `getattr(...)` only when that metadata is present
3. Treat missing metadata as "no explicit feature-stage annotation"
4. Do not rely on `ExperimentalFeature` or `ReleaseCandidateFeature` membership
over time, since staged features may move or be removed as they advance
This sample loops through the symbols exported from the root `agent_framework`
module and reports the ones that currently carry feature-stage metadata.
"""
def describe_api(name: str, api: Any) -> None:
"""Print optional feature-stage metadata for one API."""
feature_stage = getattr(api, "__feature_stage__", "released")
feature_id = getattr(api, "__feature_id__", None)
print(f"{name}:")
print(f" feature_stage = {feature_stage!r}")
print(f" feature_id = {feature_id!r}")
def iter_staged_root_exports() -> list[tuple[str, Any]]:
"""Return root exports that currently carry feature-stage metadata."""
staged_root_symbols: list[tuple[str, Any]] = []
for symbol_name in sorted(agent_framework.__all__):
symbol = getattr(agent_framework, symbol_name)
feature_stage = getattr(symbol, "__feature_stage__", None)
feature_id = getattr(symbol, "__feature_id__", None)
if feature_stage is None and feature_id is None:
continue
staged_root_symbols.append((symbol_name, symbol))
return staged_root_symbols
async def main() -> None:
"""Run the feature-stage introspection sample."""
print("Feature stage introspection")
print("-" * 60)
# 1. Loop through everything exported from the root module.
staged_root_symbols = iter_staged_root_exports()
# 2. Show the root exports that currently carry feature-stage metadata.
if not staged_root_symbols:
print("No root exports currently carry feature-stage metadata.")
return
print("Root exports with feature-stage metadata:")
for name, api in staged_root_symbols:
describe_api(name, api)
print()
print("Root exports without metadata currently have no explicit feature-stage metadata.")
if __name__ == "__main__":
asyncio.run(main())
"""
Sample output:
Feature stage introspection
------------------------------------------------------------
Root exports with feature-stage metadata:
<export name>:
feature_stage = 'experimental'
feature_id = '<feature id>'
Root exports without metadata currently have no explicit feature-stage metadata.
"""
@@ -53,3 +53,9 @@ All samples require:
- An [Azure AI Foundry](https://ai.azure.com/) project with a deployed model (e.g. `gpt-4o-mini`)
- Azure CLI authentication (`az login`)
- Environment variables set in a `.env` file (see `python/.env.example`)
## Suppressing the experimental warning
The Agent Skills APIs in these samples are still experimental. Each sample includes
a short commented `warnings.filterwarnings(...)` snippet near the imports. Uncomment
it if you want to suppress the Skills warning before using the experimental APIs.
@@ -3,6 +3,11 @@
import asyncio
import json
import os
# Uncomment this filter to suppress the experimental Skills warning before
# using the sample's Skills APIs.
# import warnings # isort: skip
# warnings.filterwarnings("ignore", message=r"\[SKILLS\].*", category=FutureWarning)
from textwrap import dedent
from typing import Any
@@ -89,7 +94,7 @@ def conversion_policy(**kwargs: Any) -> Any:
Args:
**kwargs: Runtime keyword arguments from ``agent.run()``.
For example, ``agent.run(..., precision=2)``
For example, ``agent.run(..., function_invocation_kwargs={"precision": 2})``
makes ``kwargs["precision"]`` available here.
"""
precision = kwargs.get("precision", 4)
@@ -137,15 +142,11 @@ async def main() -> None:
credential=AzureCliCredential(),
)
# Create the skills provider with the code-defined skill
skills_provider = SkillsProvider(
skills=[unit_converter_skill],
)
# Create the skills provider with the code-defined skill and pass it to the agent
async with Agent(
client=client,
instructions="You are a helpful assistant that can convert units.",
context_providers=[skills_provider],
context_providers=[SkillsProvider(skills=[unit_converter_skill])],
) as agent:
print("Converting units")
print("-" * 60)
@@ -3,6 +3,11 @@
import asyncio
import os
import sys
# 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 pathlib import Path
from agent_framework import Agent, SkillsProvider
@@ -4,6 +4,11 @@ import asyncio
import json
import os
import sys
# 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 pathlib import Path
from textwrap import dedent
from typing import Any
@@ -2,6 +2,11 @@
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, Skill, SkillsProvider
@@ -9,6 +9,11 @@ from __future__ import annotations
import subprocess
import sys
# 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 pathlib import Path
from typing import Any