# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
"""Typed Agent365 execute-tool argument and result schema models.
The models mirror the Agent365 JSON contract used by the .NET distro. Payloads are
serialized with :func:`serialize_tool_call_payload`, which never raises: any failure is
replaced by :data:`TOOL_CALL_SERIALIZATION_ERROR_JSON` so a span is never orphaned and no
caller exception escapes the telemetry path.
"""
from __future__ import annotations
import base64
import datetime
import json
import logging
import uuid
from collections.abc import Callable, Collection, Mapping
from dataclasses import dataclass, field, fields, is_dataclass
from decimal import Decimal
from enum import Enum
from typing import Any
logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())
#: Schema version emitted on every typed execute-tool payload.
TOOL_CALL_SCHEMA_VERSION = "1.0"
#: Diagnostic payload emitted when a typed execute-tool payload cannot be serialized.
TOOL_CALL_SERIALIZATION_ERROR_JSON = '{"serialization_error":"Failed to serialize execute tool payload."}'
_JSON_NAME = "json_name"
_ENUM_TYPE = "enum_type"
_EXTENSION_DATA_FIELD = "extension_data"
_EXTENSION_DATA_JSON_NAME = "metadata"
ToolCallPayload = ExecuteToolCallArguments | ExecuteToolCallResult
def _to_base64(value: bytes | bytearray | memoryview) -> str:
"""Return the base64 text for a binary value, matching the .NET byte-array contract."""
return base64.b64encode(bytes(value)).decode("ascii")
def _to_isoformat(value: datetime.datetime | datetime.date | datetime.time) -> str:
"""Return the ISO-8601 text for a date/time value."""
text = value.isoformat()
return f"{text[:-6]}Z" if text.endswith("+00:00") else text
def _preserve_decimal(value: Decimal) -> Decimal:
"""Keep a decimal value intact until JSON-number encoding."""
return value
# Ordered scalar conversions; ``bool`` must precede ``int`` and ``str`` must precede ``Collection``.
_SCALAR_CONVERTERS: tuple[tuple[type | tuple[type, ...], Callable[[Any], Any]], ...] = (
(str, str),
(bool, bool),
(int, int),
(float, float),
((bytes, bytearray, memoryview), _to_base64),
((datetime.datetime, datetime.date, datetime.time), _to_isoformat),
(uuid.UUID, str),
(Decimal, _preserve_decimal),
)
def _encode_json_value(value: Any) -> str: # pylint: disable=too-many-return-statements
"""Encode a normalized payload value, preserving decimal JSON numbers."""
if value is None:
return "null"
if isinstance(value, str):
return json.dumps(value, ensure_ascii=False)
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, int):
return str(value)
if isinstance(value, float):
return json.dumps(value, allow_nan=False)
if isinstance(value, Decimal):
if not value.is_finite():
raise ValueError("Out of range Decimal values are not JSON compliant.")
return str(value)
if isinstance(value, list):
return f"[{', '.join(_encode_json_value(item) for item in value)}]"
if isinstance(value, dict):
items = (
f"{json.dumps(key, ensure_ascii=False)}: {_encode_json_value(item_value)}"
for key, item_value in value.items()
)
return f"{{{', '.join(items)}}}"
raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable.")
def _to_json_value(value: Any, stack: set[int]) -> Any:
"""Convert a payload value into a JSON-compatible value, raising on unsupported input."""
if value is None:
return None
if isinstance(value, Enum):
return _to_json_value(value.value, stack)
for scalar_types, convert in _SCALAR_CONVERTERS:
if isinstance(value, scalar_types):
return convert(value)
if is_dataclass(value) and not isinstance(value, type):
return _dataclass_to_json_value(value, stack)
if isinstance(value, Mapping):
return _mapping_to_json_value(value, stack)
if isinstance(value, Collection):
return _collection_to_json_value(value, stack)
raise TypeError(f"Object of type {type(value).__name__} is not supported in an execute tool payload.")
def _enter(value: Any, stack: set[int]) -> int:
"""Push a container onto the active recursion path, rejecting reference cycles."""
marker = id(value)
if marker in stack:
raise ValueError("Circular reference detected in execute tool payload.")
stack.add(marker)
return marker
def _dataclass_to_json_value(value: Any, stack: set[int]) -> dict[str, Any]:
"""Serialize a schema model, omitting ``None`` properties and wrapping extension data."""
marker = _enter(value, stack)
try:
serialized: dict[str, Any] = {}
extension_data: Any = {}
for item in fields(value):
item_value = getattr(value, item.name)
if item.name == _EXTENSION_DATA_FIELD:
extension_data = item_value if item_value is not None else {}
continue
if item_value is None:
continue
json_name = item.metadata.get(_JSON_NAME, item.name)
enum_type = item.metadata.get(_ENUM_TYPE)
if enum_type is None:
serialized[json_name] = _to_json_value(item_value, stack)
else:
serialized[json_name] = _coerce_enum(item_value, enum_type, json_name)
if not isinstance(extension_data, Mapping):
raise TypeError(f"Extension data must be a mapping; got {type(extension_data).__name__}.")
if extension_data:
serialized[_EXTENSION_DATA_JSON_NAME] = _mapping_to_json_value(extension_data, stack)
return serialized
finally:
stack.discard(marker)
def _mapping_to_json_value(value: Mapping[Any, Any], stack: set[int]) -> dict[str, Any]:
"""Serialize a caller-supplied mapping, preserving ``None`` values as JSON ``null``."""
marker = _enter(value, stack)
try:
return {_json_object_key(key): _to_json_value(item_value, stack) for key, item_value in value.items()}
finally:
stack.discard(marker)
def _collection_to_json_value(value: Collection[Any], stack: set[int]) -> list[Any]:
"""Serialize a caller-supplied collection as a JSON array, preserving ``None`` items."""
marker = _enter(value, stack)
try:
return [_to_json_value(item_value, stack) for item_value in value]
finally:
stack.discard(marker)
def _json_object_key(key: Any) -> str:
"""Return the JSON object key for a mapping key, rejecting non-string keys."""
key_value = key.value if isinstance(key, Enum) else key
if not isinstance(key_value, str):
raise TypeError(f"Execute tool payload object keys must be strings; got {type(key).__name__}.")
return str(key_value)
def _coerce_enum(value: Any, enum_type: type[Enum], json_name: str) -> str:
"""Return the schema token for an enum-valued property.
Enum members are accepted, as are the exact lowercase string tokens for backward
compatibility. Undefined tokens, numeric values, booleans, and members of other
enums are rejected so schema-invalid data is never emitted.
"""
if isinstance(value, enum_type):
member_value = value.value
elif isinstance(value, str) and not isinstance(value, Enum):
try:
member_value = enum_type(value).value
except ValueError as exc:
raise ValueError(f"'{json_name}' must be one of {_allowed_enum_values(enum_type)}; got {value!r}.") from exc
else:
raise TypeError(
f"'{json_name}' must be a {enum_type.__name__} member or one of "
f"{_allowed_enum_values(enum_type)}; got {type(value).__name__}."
)
if not isinstance(member_value, str):
raise TypeError(f"'{json_name}' must serialize to a string; got {type(member_value).__name__}.")
return str(member_value)
def _allowed_enum_values(enum_type: type[Enum]) -> str:
"""Return a display list of the accepted string tokens for an enum."""
return ", ".join(repr(member.value) for member in enum_type)