agent/agent-framework
1
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: Introducing support for declarative yaml spec (#2002)

Browse Source 
* first work on declarative

* initial version of the declarative support

* fix tests and mypy

* fix parameters of functiontool

* slight logic improvement

* remove path until merge

* updates from comments

* create dispatcher and spec type, json_schema method

* fix mypy, skipping model

* updated lock

* fixed declarative tests and renamed some other test files

* refined loader

* updated lock

* fix mypy

* added readme to samples folder

* fixes from review

* undid test file rename
This commit is contained in:
Eduard van Valkenburg
2025-11-19 17:33:02 +01:00
committed by GitHub
Unverified
parent d2d0f46e15
commit 92df9e14bf
35 changed files with 3924 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/ag-ui/tests/__init__.py
-1
View File
@@ -1 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
python/packages/chatkit/tests/__init__.py
-1
View File
@@ -1 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
python/packages/core/agent_framework/_agents.py
+1 -1
View File
@@ -589,7 +589,7 @@ class ChatAgent(BaseAgent):
chat_message_store_factory: Callable[[], ChatMessageStoreProtocol] | None = None,
context_providers: ContextProvider | list[ContextProvider] | AggregateContextProvider | None = None,
middleware: Middleware | list[Middleware] | None = None,
# chat option params
# chat options
allow_multiple_tool_calls: bool | None = None,
conversation_id: str | None = None,
frequency_penalty: float | None = None,
python/packages/core/agent_framework/_clients.py
+9
View File
@@ -214,6 +214,7 @@ def _merge_chat_options(
*,
base_chat_options: ChatOptions | Any | None,
model_id: str | None = None,
allow_multiple_tool_calls: bool | None = None,
frequency_penalty: float | None = None,
logit_bias: dict[str | int, float] | None = None,
max_tokens: int | None = None,
@@ -239,6 +240,7 @@ def _merge_chat_options(
Keyword Args:
base_chat_options: Optional base ChatOptions to merge with direct parameters.
model_id: The model_id to use for the agent.
allow_multiple_tool_calls: Whether to allow multiple tool calls in a single response.
frequency_penalty: The frequency penalty to use.
logit_bias: The logit bias to use.
max_tokens: The maximum number of tokens to generate.
@@ -270,6 +272,7 @@ def _merge_chat_options(
return base_chat_options & ChatOptions(
model_id=model_id,
allow_multiple_tool_calls=allow_multiple_tool_calls,
frequency_penalty=frequency_penalty,
logit_bias=logit_bias,
max_tokens=max_tokens,
@@ -485,6 +488,7 @@ class BaseChatClient(SerializationMixin, ABC):
self,
messages: str | ChatMessage | list[str] | list[ChatMessage],
*,
allow_multiple_tool_calls: bool | None = None,
frequency_penalty: float | None = None,
logit_bias: dict[str | int, float] | None = None,
max_tokens: int | None = None,
@@ -517,6 +521,7 @@ class BaseChatClient(SerializationMixin, ABC):
messages: The message or messages to send to the model.
Keyword Args:
allow_multiple_tool_calls: Whether to allow multiple tool calls in a single response.
frequency_penalty: The frequency penalty to use.
logit_bias: The logit bias to use.
max_tokens: The maximum number of tokens to generate.
@@ -545,6 +550,7 @@ class BaseChatClient(SerializationMixin, ABC):
chat_options = _merge_chat_options(
base_chat_options=kwargs.pop("chat_options", None),
model_id=model_id,
allow_multiple_tool_calls=allow_multiple_tool_calls,
frequency_penalty=frequency_penalty,
logit_bias=logit_bias,
max_tokens=max_tokens,
@@ -580,6 +586,7 @@ class BaseChatClient(SerializationMixin, ABC):
self,
messages: str | ChatMessage | list[str] | list[ChatMessage],
*,
allow_multiple_tool_calls: bool | None = None,
frequency_penalty: float | None = None,
logit_bias: dict[str | int, float] | None = None,
max_tokens: int | None = None,
@@ -612,6 +619,7 @@ class BaseChatClient(SerializationMixin, ABC):
messages: The message or messages to send to the model.
Keyword Args:
allow_multiple_tool_calls: Whether to allow multiple tool calls in a single response.
frequency_penalty: The frequency penalty to use.
logit_bias: The logit bias to use.
max_tokens: The maximum number of tokens to generate.
@@ -640,6 +648,7 @@ class BaseChatClient(SerializationMixin, ABC):
chat_options = _merge_chat_options(
base_chat_options=kwargs.pop("chat_options", None),
model_id=model_id,
allow_multiple_tool_calls=allow_multiple_tool_calls,
frequency_penalty=frequency_penalty,
logit_bias=logit_bias,
max_tokens=max_tokens,
python/packages/core/agent_framework/declarative/__init__.py
+23
View File
@@ -0,0 +1,23 @@
# Copyright (c) Microsoft. All rights reserved.
import importlib
from typing import Any
IMPORT_PATH = "agent_framework_declarative"
PACKAGE_NAME = "agent-framework-declarative"
_IMPORTS = ["__version__", "AgentFactory", "DeclarativeLoaderError", "ProviderLookupError", "ProviderTypeMapping"]
def __getattr__(name: str) -> Any:
if name in _IMPORTS:
try:
return getattr(importlib.import_module(IMPORT_PATH), name)
except ModuleNotFoundError as exc:
raise ModuleNotFoundError(
f"The '{PACKAGE_NAME}' package is not installed, please do `pip install {PACKAGE_NAME}`"
) from exc
raise AttributeError(f"Module {IMPORT_PATH} has no attribute {name}.")
def __dir__() -> list[str]:
return _IMPORTS
python/packages/core/agent_framework/declarative/__init__.pyi
+17
View File
@@ -0,0 +1,17 @@
# Copyright (c) Microsoft. All rights reserved.
from agent_framework_declarative import (
AgentFactory,
DeclarativeLoaderError,
ProviderLookupError,
ProviderTypeMapping,
__version__,
)
__all__ = [
"AgentFactory",
"DeclarativeLoaderError",
"ProviderLookupError",
"ProviderTypeMapping",
"__version__",
]
python/packages/core/pyproject.toml
+1
View File
@@ -48,6 +48,7 @@ all = [
"agent-framework-azurefunctions",
"agent-framework-chatkit",
"agent-framework-copilotstudio",
"agent-framework-declarative",
"agent-framework-devui",
"agent-framework-lab",
"agent-framework-mem0",
python/packages/declarative/LICENSE
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) Microsoft Corporation.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE
python/packages/declarative/README.md
+11
View File
@@ -0,0 +1,11 @@
# Get Started with Microsoft Agent Framework Declarative
Please install this package via pip:
```bash
pip install agent-framework-declarative --pre
```
## Declarative features
The declarative packages provides support for building agents based on a declarative yaml specification.
python/packages/declarative/agent_framework_declarative/__init__.py
+12
View File
@@ -0,0 +1,12 @@
# Copyright (c) Microsoft. All rights reserved.
from importlib import metadata
from ._loader import AgentFactory, DeclarativeLoaderError, ProviderLookupError, ProviderTypeMapping
try:
__version__ = metadata.version(__name__)
except metadata.PackageNotFoundError:
__version__ = "0.0.0" # Fallback for development mode
__all__ = ["AgentFactory", "DeclarativeLoaderError", "ProviderLookupError", "ProviderTypeMapping", "__version__"]
python/packages/declarative/agent_framework_declarative/_loader.py
+422
View File
@@ -0,0 +1,422 @@
# Copyright (c) Microsoft. All rights reserved.
from collections.abc import Callable, Mapping
from pathlib import Path
from typing import Any, Literal, TypedDict
import yaml
from agent_framework import (
AIFunction,
ChatAgent,
ChatClientProtocol,
HostedCodeInterpreterTool,
HostedFileContent,
HostedFileSearchTool,
HostedMCPSpecificApproval,
HostedMCPTool,
HostedVectorStoreContent,
HostedWebSearchTool,
ToolProtocol,
)
from agent_framework._tools import _create_model_from_json_schema # type: ignore
from agent_framework.exceptions import AgentFrameworkException
from dotenv import load_dotenv
from ._models import (
AnonymousConnection,
ApiKeyConnection,
CodeInterpreterTool,
FileSearchTool,
FunctionTool,
McpServerToolSpecifyApprovalMode,
McpTool,
Model,
ModelOptions,
PromptAgent,
ReferenceConnection,
RemoteConnection,
Tool,
WebSearchTool,
agent_schema_dispatch,
)
class ProviderTypeMapping(TypedDict, total=True):
package: str
name: str
model_id_field: str
PROVIDER_TYPE_OBJECT_MAPPING: dict[str, ProviderTypeMapping] = {
"AzureOpenAI.Chat": {
"package": "agent_framework.azure",
"name": "AzureOpenAIChatClient",
"model_id_field": "deployment_name",
},
"AzureOpenAI.Assistants": {
"package": "agent_framework.azure",
"name": "AzureOpenAIAssistantsClient",
"model_id_field": "deployment_name",
},
"AzureOpenAI.Responses": {
"package": "agent_framework.azure",
"name": "AzureOpenAIResponsesClient",
"model_id_field": "deployment_name",
},
"OpenAI.Chat": {
"package": "agent_framework.openai",
"name": "OpenAIChatClient",
"model_id_field": "model_id",
},
"OpenAI.Assistants": {
"package": "agent_framework.openai",
"name": "OpenAIAssistantsClient",
"model_id_field": "model_id",
},
"OpenAI.Responses": {
"package": "agent_framework.openai",
"name": "OpenAIResponsesClient",
"model_id_field": "model_id",
},
"AzureAIAgentClient": {
"package": "agent_framework.azure",
"name": "AzureAIAgentClient",
"model_id_field": "model_deployment_name",
},
"AzureAIClient": {
"package": "agent_framework.azure",
"name": "AzureAIClient",
"model_id_field": "model_deployment_name",
},
"Anthropic.Chat": {
"package": "agent_framework.anthropic",
"name": "AnthropicChatClient",
"model_id_field": "model_id",
},
}
class DeclarativeLoaderError(AgentFrameworkException):
"""Exception raised for errors in the declarative loader."""
pass
class ProviderLookupError(DeclarativeLoaderError):
"""Exception raised for errors in provider type lookup."""
pass
class AgentFactory:
def __init__(
self,
*,
chat_client: ChatClientProtocol | None = None,
bindings: Mapping[str, Any] | None = None,
connections: Mapping[str, Any] | None = None,
client_kwargs: Mapping[str, Any] | None = None,
additional_mappings: Mapping[str, ProviderTypeMapping] | None = None,
default_provider: str = "AzureAIClient",
env_file: str | None = None,
) -> None:
"""Create the agent factory, with bindings.
Args:
chat_client: An optional ChatClientProtocol instance to use as a dependency,
this will be passed to the ChatAgent that get's created.
If you need to create multiple agents with different chat clients,
do not pass this and instead provide the chat client in the YAML definition.
bindings: An optional dictionary of bindings to use when creating agents.
connections: An optional dictionary of connections to resolve ReferenceConnections.
client_kwargs: An optional dictionary of keyword arguments to pass to chat client constructor.
additional_mappings: An optional dictionary to extend the provider type to object mapping.
Should have the structure:
..code-block:: python
additional_mappings = {
"Provider.ApiType": {
"package": "package.name",
"name": "ClassName",
"model_id_field": "field_name_in_constructor",
},
...
}
Here, "Provider.ApiType" is the lookup key used when both provider and apiType are specified in the
model, "Provider" is also allowed.
Package refers to which model needs to be imported, Name is the class name of the ChatClientProtocol
implementation, and model_id_field is the name of the field in the constructor
that accepts the model.id value.
default_provider: The default provider used when model.provider is not specified,
default is "AzureAIClient".
env_file: An optional path to a .env file to load environment variables from.
"""
self.chat_client = chat_client
self.bindings = bindings
self.connections = connections
self.client_kwargs = client_kwargs or {}
self.additional_mappings = additional_mappings or {}
self.default_provider: str = default_provider
load_dotenv(dotenv_path=env_file)
def create_agent_from_yaml_path(self, yaml_path: str | Path) -> ChatAgent:
"""Create a ChatAgent from a YAML file path.
This method does the following things:
1. Loads the YAML file into a AgentSchema object using open and agent_schema_dispatch.
2. Validates that the loaded object is a PromptAgent.
3. Creates the appropriate ChatClient based on the model provider and apiType.
4. Parses the tools, options, and response format from the PromptAgent.
5. Creates and returns a ChatAgent instance with the configured properties.
Args:
yaml_path: Path to the YAML file representation of a AgentSchema object
Returns:
The ``ChatAgent`` instance created from the YAML file.
Raises:
DeclarativeLoaderError: If the YAML does not represent a PromptAgent.
ProviderLookupError: If the provider type is unknown or unsupported.
ValueError: If a ReferenceConnection cannot be resolved.
ModuleNotFoundError: If the required module for the provider type cannot be imported.
AttributeError: If the required class for the provider type cannot be found in the module.
"""
if not isinstance(yaml_path, Path):
yaml_path = Path(yaml_path)
if not yaml_path.exists():
raise DeclarativeLoaderError(f"YAML file not found at path: {yaml_path}")
with open(yaml_path) as f:
yaml_str = f.read()
return self.create_agent_from_yaml(yaml_str)
def create_agent_from_yaml(self, yaml_str: str) -> ChatAgent:
"""Create a ChatAgent from a YAML string.
This method does the following things:
1. Loads the YAML string into a AgentSchema object using agent_schema_dispatch.
2. Validates that the loaded object is a PromptAgent.
3. Creates the appropriate ChatClient based on the model provider and apiType.
4. Parses the tools, options, and response format from the PromptAgent.
5. Creates and returns a ChatAgent instance with the configured properties.
Args:
yaml_str: YAML string representation of a AgentSchema object
Returns:
The ``ChatAgent`` instance created from the YAML string.
Raises:
DeclarativeLoaderError: If the YAML does not represent a PromptAgent.
ProviderLookupError: If the provider type is unknown or unsupported.
ValueError: If a ReferenceConnection cannot be resolved.
ModuleNotFoundError: If the required module for the provider type cannot be imported.
AttributeError: If the required class for the provider type cannot be found in the module.
"""
prompt_agent = agent_schema_dispatch(yaml.safe_load(yaml_str))
if not isinstance(prompt_agent, PromptAgent):
raise DeclarativeLoaderError("Only yaml definitions for a PromptAgent are supported for agent creation.")
# Step 1: Create the ChatClient
client = self._get_client(prompt_agent)
# Step 2: Get the chat options
chat_options = self._parse_chat_options(prompt_agent.model)
if tools := self._parse_tools(prompt_agent.tools):
chat_options["tools"] = tools
if output_schema := prompt_agent.outputSchema:
chat_options["response_format"] = _create_model_from_json_schema("agent", output_schema.to_json_schema())
# Step 3: Create the agent instance
return ChatAgent(
chat_client=client,
name=prompt_agent.name,
description=prompt_agent.description,
instructions=prompt_agent.instructions,
**chat_options,
)
def _get_client(self, prompt_agent: PromptAgent) -> ChatClientProtocol:
"""Create the ChatClientProtocol instance based on the PromptAgent model."""
if not prompt_agent.model:
# if no model is defined, use the supplied chat_client
if self.chat_client:
return self.chat_client
raise DeclarativeLoaderError(
"ChatClient must be provided to create agent from PromptAgent, "
"alternatively define a model in the PromptAgent."
)
setup_dict: dict[str, Any] = {}
setup_dict.update(self.client_kwargs)
# parse connections
if prompt_agent.model.connection:
match prompt_agent.model.connection:
case ApiKeyConnection():
setup_dict["api_key"] = prompt_agent.model.connection.apiKey
if prompt_agent.model.connection.endpoint:
setup_dict["endpoint"] = prompt_agent.model.connection.endpoint
case RemoteConnection() | AnonymousConnection():
setup_dict["endpoint"] = prompt_agent.model.connection.endpoint
case ReferenceConnection():
if not self.connections:
raise ValueError("Connections must be provided to resolve ReferenceConnection")
# find the referenced connection
if prompt_agent.model.connection.name and (
value := self.connections.get(prompt_agent.model.connection.name)
):
setup_dict[prompt_agent.model.connection.name] = value
else:
raise ValueError(
f"ReferenceConnection with name {prompt_agent.model.connection.name} not found in provided "
"connections."
)
# Any client we create, needs a model.id
if not prompt_agent.model.id:
# if prompt_agent.model is defined, but no id, use the supplied chat_client
if self.chat_client:
return self.chat_client
# or raise, since we cannot create a client without model id
raise DeclarativeLoaderError(
"ChatClient must be provided to create agent from PromptAgent, or define model.id in the PromptAgent."
)
# if provider is defined, use that, if possible with apiType, fallback to default_provider
mapping = self._retrieve_provider_configuration(prompt_agent.model)
module_name = mapping["package"]
class_name = mapping["name"]
module = __import__(module_name, fromlist=[class_name])
agent_class = getattr(module, class_name)
setup_dict[mapping["model_id_field"]] = prompt_agent.model.id
return agent_class(**setup_dict) # type: ignore[no-any-return]
def _parse_chat_options(self, model: Model | None) -> dict[str, Any]:
"""Parse ModelOptions into chat options dictionary."""
chat_options: dict[str, Any] = {}
if not model or not model.options or not isinstance(model.options, ModelOptions):
return chat_options
options = model.options
if options.frequencyPenalty is not None:
chat_options["frequency_penalty"] = options.frequencyPenalty
if options.presencePenalty is not None:
chat_options["presence_penalty"] = options.presencePenalty
if options.maxOutputTokens is not None:
chat_options["max_tokens"] = options.maxOutputTokens
if options.temperature is not None:
chat_options["temperature"] = options.temperature
if options.topP is not None:
chat_options["top_p"] = options.topP
if options.seed is not None:
chat_options["seed"] = options.seed
if options.stopSequences:
chat_options["stop"] = options.stopSequences
if options.allowMultipleToolCalls is not None:
chat_options["allow_multiple_tool_calls"] = options.allowMultipleToolCalls
if (chat_tool_mode := options.additionalProperties.pop("chatToolMode", None)) is not None:
chat_options["tool_choice"] = chat_tool_mode
if options.additionalProperties:
chat_options["additional_chat_options"] = options.additionalProperties
return chat_options
def _parse_tools(self, tools: list[Tool] | None) -> list[ToolProtocol] | None:
"""Parse tool resources into ToolProtocol instances."""
if not tools:
return None
return [self._parse_tool(tool_resource) for tool_resource in tools]
def _parse_tool(self, tool_resource: Tool) -> ToolProtocol:
"""Parse a single tool resource into a ToolProtocol instance."""
match tool_resource:
case FunctionTool():
func: Callable[..., Any] | None = None
if self.bindings and tool_resource.bindings:
for binding in tool_resource.bindings:
if binding.name and (func := self.bindings.get(binding.name)):
break
return AIFunction( # type: ignore
name=tool_resource.name, # type: ignore
description=tool_resource.description, # type: ignore
input_model=tool_resource.parameters.to_json_schema() if tool_resource.parameters else None,
func=func,
)
case WebSearchTool():
return HostedWebSearchTool(
description=tool_resource.description, additional_properties=tool_resource.options
)
case FileSearchTool():
add_props: dict[str, Any] = {}
if tool_resource.ranker is not None:
add_props["ranker"] = tool_resource.ranker
if tool_resource.scoreThreshold is not None:
add_props["score_threshold"] = tool_resource.scoreThreshold
if tool_resource.filters:
add_props["filters"] = tool_resource.filters
return HostedFileSearchTool(
inputs=[HostedVectorStoreContent(id) for id in tool_resource.vectorStoreIds or []],
description=tool_resource.description,
max_results=tool_resource.maximumResultCount,
additional_properties=add_props,
)
case CodeInterpreterTool():
return HostedCodeInterpreterTool(
inputs=[HostedFileContent(file_id=file) for file in tool_resource.fileIds or []],
description=tool_resource.description,
)
case McpTool():
approval_mode: HostedMCPSpecificApproval | Literal["always_require", "never_require"] | None = None
if tool_resource.approvalMode is not None:
if tool_resource.approvalMode.kind == "always":
approval_mode = "always_require"
elif tool_resource.approvalMode.kind == "never":
approval_mode = "never_require"
elif isinstance(tool_resource.approvalMode, McpServerToolSpecifyApprovalMode):
approval_mode = {}
if tool_resource.approvalMode.alwaysRequireApprovalTools:
approval_mode["always_require_approval"] = (
tool_resource.approvalMode.alwaysRequireApprovalTools
)
if tool_resource.approvalMode.neverRequireApprovalTools:
approval_mode["never_require_approval"] = (
tool_resource.approvalMode.neverRequireApprovalTools
)
if not approval_mode:
approval_mode = None
return HostedMCPTool(
name=tool_resource.name, # type: ignore
description=tool_resource.description,
url=tool_resource.url, # type: ignore
allowed_tools=tool_resource.allowedTools,
approval_mode=approval_mode,
)
case _:
raise ValueError(f"Unsupported tool kind: {tool_resource.kind}")
def _retrieve_provider_configuration(self, model: Model) -> ProviderTypeMapping:
"""Retrieve the provider configuration based on the model's provider and apiType.
If only provider is specified, it will be used.
If both provider and apiType are specified, both will be used.
If neither is specified, the default_provider will be used.
Args:
model: The Model instance containing provider and apiType information.
Returns:
A dictionary containing the package, name, and model_id_field for the provider.
Raises:
ProviderLookupError: If the provider type is not supported or can't be found.
"""
class_lookup = (
f"{model.provider}.{model.apiType}"
if model.apiType
else f"{model.provider}"
if model.provider
else self.default_provider
)
if class_lookup in self.additional_mappings:
return self.additional_mappings[class_lookup]
if class_lookup not in PROVIDER_TYPE_OBJECT_MAPPING:
raise ProviderLookupError(f"Unsupported provider type: {class_lookup}")
return PROVIDER_TYPE_OBJECT_MAPPING[class_lookup]
python/packages/declarative/agent_framework_declarative/_models.py
+1093
View File
File diff suppressed because it is too large Load Diff
python/packages/declarative/pyproject.toml
+97
View File
@@ -0,0 +1,97 @@
[project]
name = "agent-framework-declarative"
description = "Declarative specification support for Microsoft Agent Framework."
authors = [{ name = "Microsoft", email = "af-support@microsoft.com"}]
readme = "README.md"
requires-python = ">=3.10"
version = "1.0.0b251028"
license-files = ["LICENSE"]
urls.homepage = "https://aka.ms/agent-framework"
urls.source = "https://github.com/microsoft/agent-framework/tree/main/python"
urls.release_notes = "https://github.com/microsoft/agent-framework/releases?q=tag%3Apython-1&expanded=true"
urls.issues = "https://github.com/microsoft/agent-framework/issues"
classifiers = [
"License :: OSI Approved :: MIT License",
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Typing :: Typed",
]
dependencies = [
"agent-framework-core",
"powerfx>=0.0.31; python_version < '3.14'",
"pyyaml>=6.0,<7.0",
]
[dependency-groups]
dev = [
"types-PyYaml"
]
[tool.uv]
prerelease = "if-necessary-or-explicit"
environments = [
"sys_platform == 'darwin'",
"sys_platform == 'linux'",
"sys_platform == 'win32'"
]
[tool.uv-dynamic-versioning]
fallback-version = "0.0.0"
[tool.pytest.ini_options]
testpaths = 'tests'
addopts = "-ra -q -r fEX"
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
filterwarnings = [
"ignore:Support for class-based `config` is deprecated:DeprecationWarning:pydantic.*"
]
timeout = 120
[tool.ruff]
extend = "../../pyproject.toml"
[tool.coverage.run]
omit = [
"**/__init__.py"
]
[tool.pyright]
extends = "../../pyproject.toml"
exclude = ['tests']
[tool.mypy]
plugins = ['pydantic.mypy']
strict = true
python_version = "3.10"
ignore_missing_imports = true
disallow_untyped_defs = true
no_implicit_optional = true
check_untyped_defs = true
warn_return_any = true
show_error_codes = true
warn_unused_ignores = false
disallow_incomplete_defs = true
disallow_untyped_decorators = true
exclude = [
'_models.py$',
]
[tool.bandit]
targets = ["agent_framework_declarative"]
exclude_dirs = ["tests"]
[tool.poe]
executor.type = "uv"
include = "../../shared_tasks.toml"
[tool.poe.tasks]
mypy = "mypy --config-file $POE_ROOT/pyproject.toml agent_framework_declarative"
test = "pytest --cov=agent_framework_declarative --cov-report=term-missing:skip-covered tests"
[build-system]
requires = ["flit-core >= 3.11,<4.0"]
build-backend = "flit_core.buildapi"
python/packages/declarative/tests/test_declarative_loader.py
+456
View File
@@ -0,0 +1,456 @@
# Copyright (c) Microsoft. All rights reserved.
import sys
from pathlib import Path
from typing import Any
import pytest
import yaml
from agent_framework_declarative._models import (
AgentDefinition,
AgentManifest,
AnonymousConnection,
ApiKeyConnection,
ArrayProperty,
CodeInterpreterTool,
Connection,
CustomTool,
FileSearchTool,
FunctionTool,
McpServerApprovalMode,
McpServerToolAlwaysRequireApprovalMode,
McpServerToolNeverRequireApprovalMode,
McpServerToolSpecifyApprovalMode,
McpTool,
ModelResource,
ObjectProperty,
OpenApiTool,
PromptAgent,
Property,
PropertySchema,
ReferenceConnection,
RemoteConnection,
Resource,
ToolResource,
WebSearchTool,
agent_schema_dispatch,
)
pytestmark = pytest.mark.skipif(sys.version_info >= (3, 14), reason="Skipping on Python 3.14+")
@pytest.mark.parametrize(
"yaml_content,expected_type,expected_attributes",
[
# Agent Manifest (no kind field)
(
"""
name: my-manifest
description: A test manifest
""",
AgentManifest,
{"name": "my-manifest", "description": "A test manifest"},
),
# PromptAgent
(
"""
kind: Prompt
name: assistant
description: A helpful assistant
model:
id: gpt-4
""",
PromptAgent,
{"name": "assistant", "description": "A helpful assistant"},
),
# AgentDefinition
(
"""
kind: Agent
name: base-agent
description: A base agent
""",
AgentDefinition,
{"name": "base-agent", "description": "A base agent"},
),
# ModelResource
(
"""
kind: Model
name: my-model
id: gpt-4
""",
ModelResource,
{"name": "my-model", "id": "gpt-4"},
),
# ToolResource
(
"""
kind: Tool
name: my-tool
id: search-tool
""",
ToolResource,
{"name": "my-tool", "id": "search-tool"},
),
# Resource (base)
(
"""
kind: Resource
name: generic-resource
""",
Resource,
{"name": "generic-resource"},
),
# FunctionTool
(
"""
kind: function
name: get_weather
description: Get the weather
""",
FunctionTool,
{"name": "get_weather", "description": "Get the weather"},
),
# CustomTool
(
"""
kind: custom
name: custom_tool
description: A custom tool
""",
CustomTool,
{"name": "custom_tool", "description": "A custom tool"},
),
# WebSearchTool
(
"""
kind: web_search
name: search
description: Search the web
""",
WebSearchTool,
{"name": "search", "description": "Search the web"},
),
# FileSearchTool
(
"""
kind: file_search
name: file_search
description: Search files
""",
FileSearchTool,
{"name": "file_search", "description": "Search files"},
),
# McpTool
(
"""
kind: mcp
name: mcp_tool
description: An MCP tool
serverName: my-server
""",
McpTool,
{"name": "mcp_tool", "serverName": "my-server"},
),
# OpenApiTool
(
"""
kind: openapi
name: api_tool
description: An OpenAPI tool
specification: https://api.example.com/openapi.json
""",
OpenApiTool,
{"name": "api_tool", "specification": "https://api.example.com/openapi.json"},
),
# CodeInterpreterTool
(
"""
kind: code_interpreter
name: code_tool
description: A code interpreter tool
""",
CodeInterpreterTool,
{"name": "code_tool", "description": "A code interpreter tool"},
),
# ReferenceConnection
(
"""
kind: reference
name: my-connection
target: target-connection
""",
ReferenceConnection,
{"name": "my-connection", "target": "target-connection"},
),
# RemoteConnection
(
"""
kind: remote
endpoint: https://api.example.com
""",
RemoteConnection,
{"endpoint": "https://api.example.com"},
),
# ApiKeyConnection
(
"""
kind: key
apiKey: secret-key
endpoint: https://api.example.com
""",
ApiKeyConnection,
{"apiKey": "secret-key", "endpoint": "https://api.example.com"},
),
# AnonymousConnection
(
"""
kind: anonymous
endpoint: https://api.example.com
""",
AnonymousConnection,
{"endpoint": "https://api.example.com"},
),
# Connection (base)
(
"""
kind: connection
authenticationMode: oauth
""",
Connection,
{"authenticationMode": "oauth"},
),
# ArrayProperty
(
"""
kind: array
name: items
description: An array of items
""",
ArrayProperty,
{"name": "items", "description": "An array of items"},
),
# ObjectProperty
(
"""
kind: object
name: config
description: Configuration object
""",
ObjectProperty,
{"name": "config", "description": "Configuration object"},
),
# Property (base)
(
"""
kind: property
name: field
description: A property field
""",
Property,
{"name": "field", "description": "A property field"},
),
# McpServerToolAlwaysRequireApprovalMode
(
"""
kind: always
""",
McpServerToolAlwaysRequireApprovalMode,
{},
),
# McpServerToolNeverRequireApprovalMode
(
"""
kind: never
""",
McpServerToolNeverRequireApprovalMode,
{},
),
# McpServerToolSpecifyApprovalMode
(
"""
kind: specify
alwaysRequireApprovalTools: []
neverRequireApprovalTools: []
""",
McpServerToolSpecifyApprovalMode,
{},
),
# McpServerApprovalMode (base)
(
"""
kind: approval_mode
""",
McpServerApprovalMode,
{},
),
],
)
def test_agent_schema_dispatch_all_types(yaml_content: str, expected_type: type, expected_attributes: dict[str, Any]):
"""Test that agent_schema_dispatch correctly loads all MAML object types."""
result = agent_schema_dispatch(yaml.safe_load(yaml_content))
# Check the type is correct
assert isinstance(result, expected_type), f"Expected {expected_type.__name__}, got {type(result).__name__}"
# Check expected attributes
for attr_name, attr_value in expected_attributes.items():
assert hasattr(result, attr_name), f"Result missing attribute '{attr_name}'"
assert getattr(result, attr_name) == attr_value, (
f"Attribute '{attr_name}' has value {getattr(result, attr_name)}, expected {attr_value}"
)
def test_agent_schema_dispatch_unknown_kind():
"""Test that agent_schema_dispatch returns None for unknown kind."""
yaml_content = """
kind: unknown_type
name: test
"""
result = agent_schema_dispatch(yaml.safe_load(yaml_content))
assert result is None
def test_agent_schema_dispatch_complex_agent_manifest():
"""Test loading a complex agent manifest with nested objects."""
yaml_content = """
name: complex-manifest
description: A complete manifest
template:
kind: Prompt
name: assistant
description: A helpful assistant
model:
id: gpt-4
provider: openai
tools:
- kind: web_search
name: search
description: Search the web
- kind: function
name: calculator
description: Calculate math
resources:
- kind: model
name: model1
id: gpt-4
- kind: tool
name: tool1
id: search
"""
result = agent_schema_dispatch(yaml.safe_load(yaml_content))
assert isinstance(result, AgentManifest)
assert result.name == "complex-manifest"
assert result.description == "A complete manifest"
assert isinstance(result.template, PromptAgent)
assert result.template.name == "assistant"
assert len(result.resources) == 2
assert isinstance(result.resources[0], ModelResource)
assert isinstance(result.resources[1], ToolResource)
def test_agent_schema_dispatch_prompt_agent_with_tools():
"""Test loading a prompt agent with multiple tools."""
yaml_content = """
kind: Prompt
name: multi-tool-agent
description: Agent with multiple tools
model:
id: gpt-4
tools:
- kind: web_search
name: search
description: Search the web
- kind: function
name: get_weather
description: Get weather information
- kind: code_interpreter
name: code
description: Execute code
"""
result = agent_schema_dispatch(yaml.safe_load(yaml_content))
assert isinstance(result, PromptAgent)
assert result.name == "multi-tool-agent"
assert len(result.tools) == 3
# Tools are polymorphically created based on their kind
assert result.tools[0].kind == "web_search"
assert result.tools[1].kind == "function"
assert result.tools[2].kind == "code_interpreter"
def test_agent_schema_dispatch_model_resource():
"""Test loading a model resource."""
yaml_content = """
kind: Model
name: my-model
id: gpt-4
"""
result = agent_schema_dispatch(yaml.safe_load(yaml_content))
assert isinstance(result, ModelResource)
assert result.id == "gpt-4"
def test_agent_schema_dispatch_property_schema_with_nested_properties():
"""Test loading a property schema with nested properties."""
yaml_content = """
kind: property_schema
strict: true
properties:
- kind: property
name: name
description: User name
- kind: object
name: address
description: User address
properties:
- kind: property
name: street
description: Street address
- kind: property
name: city
description: City name
- kind: array
name: tags
description: User tags
"""
result = agent_schema_dispatch(yaml.safe_load(yaml_content))
assert isinstance(result, PropertySchema)
assert result.strict is True
assert len(result.properties) == 3
# Properties are polymorphically created based on their kind
assert result.properties[0].kind == "property"
assert result.properties[1].kind == "object"
assert result.properties[2].kind == "array"
def _get_agent_sample_yaml_files() -> list[tuple[Path, Path]]:
"""Helper function to collect all YAML files from agent-samples directory."""
current_file = Path(__file__)
repo_root = current_file.parent.parent.parent.parent # tests -> declarative -> packages -> python
agent_samples_dir = repo_root.parent / "agent-samples"
if not agent_samples_dir.exists():
return []
yaml_files = list(agent_samples_dir.rglob("*.yaml")) + list(agent_samples_dir.rglob("*.yml"))
return [(yaml_file, agent_samples_dir) for yaml_file in yaml_files]
@pytest.mark.parametrize(
"yaml_file,agent_samples_dir",
_get_agent_sample_yaml_files(),
ids=lambda x: x[0].name if isinstance(x, tuple) else str(x),
)
def test_agent_schema_dispatch_agent_samples(yaml_file: Path, agent_samples_dir: Path):
"""Test that agent_schema_dispatch successfully loads a YAML file from agent-samples directory."""
with open(yaml_file) as f:
content = f.read()
result = agent_schema_dispatch(yaml.safe_load(content))
# Result can be None for unknown kinds, but should not raise exceptions
assert result is not None, f"agent_schema_dispatch returned None for {yaml_file.relative_to(agent_samples_dir)}"
python/packages/declarative/tests/test_declarative_models.py
+1049
View File
File diff suppressed because it is too large Load Diff
python/packages/purview/tests/test_models.py → python/packages/purview/tests/test_purview_models.py
View File