mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Add Python parity for HttpRequestAction in declarative workflow (#5599)
* Add Python parity for HttpRequestAction in declarative workflow * Ran pyupgrade and pright to fix CI issues * Fix conversation ID dot parsing for http executor * Removed unnecessary export command
This commit is contained in:
committed by
GitHub
Unverified
parent
18293ffb31
commit
bc42874690
@@ -6,9 +6,14 @@ from ._loader import AgentFactory, DeclarativeLoaderError, ProviderLookupError,
|
||||
from ._workflows import (
|
||||
AgentExternalInputRequest,
|
||||
AgentExternalInputResponse,
|
||||
DeclarativeActionError,
|
||||
DeclarativeWorkflowError,
|
||||
DefaultHttpRequestHandler,
|
||||
ExternalInputRequest,
|
||||
ExternalInputResponse,
|
||||
HttpRequestHandler,
|
||||
HttpRequestInfo,
|
||||
HttpRequestResult,
|
||||
WorkflowFactory,
|
||||
WorkflowState,
|
||||
)
|
||||
@@ -22,10 +27,15 @@ __all__ = [
|
||||
"AgentExternalInputRequest",
|
||||
"AgentExternalInputResponse",
|
||||
"AgentFactory",
|
||||
"DeclarativeActionError",
|
||||
"DeclarativeLoaderError",
|
||||
"DeclarativeWorkflowError",
|
||||
"DefaultHttpRequestHandler",
|
||||
"ExternalInputRequest",
|
||||
"ExternalInputResponse",
|
||||
"HttpRequestHandler",
|
||||
"HttpRequestInfo",
|
||||
"HttpRequestResult",
|
||||
"ProviderLookupError",
|
||||
"ProviderTypeMapping",
|
||||
"WorkflowFactory",
|
||||
|
||||
@@ -25,6 +25,7 @@ from ._declarative_base import (
|
||||
LoopIterationResult,
|
||||
)
|
||||
from ._declarative_builder import ALL_ACTION_EXECUTORS, DeclarativeWorkflowBuilder
|
||||
from ._errors import DeclarativeActionError, DeclarativeWorkflowError
|
||||
from ._executors_agents import (
|
||||
AGENT_ACTION_EXECUTORS,
|
||||
AGENT_REGISTRY_KEY,
|
||||
@@ -67,6 +68,10 @@ from ._executors_external_input import (
|
||||
RequestExternalInputExecutor,
|
||||
WaitForInputExecutor,
|
||||
)
|
||||
from ._executors_http import (
|
||||
HTTP_ACTION_EXECUTORS,
|
||||
HttpRequestActionExecutor,
|
||||
)
|
||||
from ._executors_tools import (
|
||||
FUNCTION_TOOL_REGISTRY_KEY,
|
||||
TOOL_ACTION_EXECUTORS,
|
||||
@@ -78,7 +83,13 @@ from ._executors_tools import (
|
||||
ToolApprovalState,
|
||||
ToolInvocationResult,
|
||||
)
|
||||
from ._factory import DeclarativeWorkflowError, WorkflowFactory
|
||||
from ._factory import WorkflowFactory
|
||||
from ._http_handler import (
|
||||
DefaultHttpRequestHandler,
|
||||
HttpRequestHandler,
|
||||
HttpRequestInfo,
|
||||
HttpRequestResult,
|
||||
)
|
||||
from ._state import WorkflowState
|
||||
|
||||
__all__ = [
|
||||
@@ -90,6 +101,7 @@ __all__ = [
|
||||
"DECLARATIVE_STATE_KEY",
|
||||
"EXTERNAL_INPUT_EXECUTORS",
|
||||
"FUNCTION_TOOL_REGISTRY_KEY",
|
||||
"HTTP_ACTION_EXECUTORS",
|
||||
"TOOL_ACTION_EXECUTORS",
|
||||
"TOOL_APPROVAL_STATE_KEY",
|
||||
"TOOL_REGISTRY_KEY",
|
||||
@@ -106,12 +118,14 @@ __all__ = [
|
||||
"ContinueLoopExecutor",
|
||||
"ConversationData",
|
||||
"CreateConversationExecutor",
|
||||
"DeclarativeActionError",
|
||||
"DeclarativeActionExecutor",
|
||||
"DeclarativeMessage",
|
||||
"DeclarativeStateData",
|
||||
"DeclarativeWorkflowBuilder",
|
||||
"DeclarativeWorkflowError",
|
||||
"DeclarativeWorkflowState",
|
||||
"DefaultHttpRequestHandler",
|
||||
"EmitEventExecutor",
|
||||
"EndConversationExecutor",
|
||||
"EndWorkflowExecutor",
|
||||
@@ -120,6 +134,10 @@ __all__ = [
|
||||
"ExternalLoopState",
|
||||
"ForeachInitExecutor",
|
||||
"ForeachNextExecutor",
|
||||
"HttpRequestActionExecutor",
|
||||
"HttpRequestHandler",
|
||||
"HttpRequestInfo",
|
||||
"HttpRequestResult",
|
||||
"InvokeAzureAgentExecutor",
|
||||
"InvokeFunctionToolExecutor",
|
||||
"JoinExecutor",
|
||||
|
||||
+23
@@ -26,6 +26,7 @@ from ._declarative_base import (
|
||||
DeclarativeActionExecutor,
|
||||
LoopIterationResult,
|
||||
)
|
||||
from ._errors import DeclarativeWorkflowError
|
||||
from ._executors_agents import AGENT_ACTION_EXECUTORS, InvokeAzureAgentExecutor
|
||||
from ._executors_basic import BASIC_ACTION_EXECUTORS
|
||||
from ._executors_control_flow import (
|
||||
@@ -39,7 +40,9 @@ from ._executors_control_flow import (
|
||||
SwitchEvaluatorExecutor,
|
||||
)
|
||||
from ._executors_external_input import EXTERNAL_INPUT_EXECUTORS
|
||||
from ._executors_http import HTTP_ACTION_EXECUTORS, HttpRequestActionExecutor
|
||||
from ._executors_tools import TOOL_ACTION_EXECUTORS, InvokeFunctionToolExecutor
|
||||
from ._http_handler import HttpRequestHandler
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -51,6 +54,7 @@ ALL_ACTION_EXECUTORS = {
|
||||
**AGENT_ACTION_EXECUTORS,
|
||||
**EXTERNAL_INPUT_EXECUTORS,
|
||||
**TOOL_ACTION_EXECUTORS,
|
||||
**HTTP_ACTION_EXECUTORS,
|
||||
}
|
||||
|
||||
# Action kinds that terminate control flow (no fall-through to successor)
|
||||
@@ -85,6 +89,7 @@ ACTION_REQUIRED_FIELDS: dict[str, list[str]] = {
|
||||
"WaitForHumanInput": ["variable"],
|
||||
"EmitEvent": ["event"],
|
||||
"InvokeFunctionTool": ["functionName"],
|
||||
"HttpRequestAction": ["url"],
|
||||
}
|
||||
|
||||
# Alternate field names that satisfy required field requirements
|
||||
@@ -129,6 +134,7 @@ class DeclarativeWorkflowBuilder:
|
||||
checkpoint_storage: Any | None = None,
|
||||
validate: bool = True,
|
||||
max_iterations: int | None = None,
|
||||
http_request_handler: HttpRequestHandler | None = None,
|
||||
):
|
||||
"""Initialize the builder.
|
||||
|
||||
@@ -141,6 +147,9 @@ class DeclarativeWorkflowBuilder:
|
||||
validate: Whether to validate the workflow definition before building (default: True)
|
||||
max_iterations: Maximum runner supersteps. Falls back to the YAML ``maxTurns``
|
||||
field, then to the core default (100).
|
||||
http_request_handler: Handler used to dispatch HttpRequestAction requests.
|
||||
Must be supplied when the workflow contains any HttpRequestAction;
|
||||
otherwise build raises ``DeclarativeWorkflowError``.
|
||||
"""
|
||||
self._yaml_def = yaml_definition
|
||||
self._workflow_id = workflow_id or yaml_definition.get("name", "declarative_workflow")
|
||||
@@ -152,6 +161,7 @@ class DeclarativeWorkflowBuilder:
|
||||
self._pending_gotos: list[tuple[Any, str]] = [] # (goto_executor, target_id)
|
||||
self._validate = validate
|
||||
self._seen_explicit_ids: set[str] = set() # Track explicit IDs for duplicate detection
|
||||
self._http_request_handler = http_request_handler
|
||||
# Resolve max_iterations: explicit arg > YAML maxTurns > core default
|
||||
resolved = max_iterations if max_iterations is not None else yaml_definition.get("maxTurns")
|
||||
if resolved is not None and (not isinstance(resolved, int) or resolved <= 0):
|
||||
@@ -458,6 +468,19 @@ class DeclarativeWorkflowBuilder:
|
||||
executor = InvokeAzureAgentExecutor(action_def, id=action_id, agents=self._agents)
|
||||
elif kind == "InvokeFunctionTool":
|
||||
executor = InvokeFunctionToolExecutor(action_def, id=action_id, tools=self._tools)
|
||||
elif kind == "HttpRequestAction":
|
||||
if self._http_request_handler is None:
|
||||
raise DeclarativeWorkflowError(
|
||||
f"Workflow defines HttpRequestAction '{action_id}' but no "
|
||||
"http_request_handler was supplied to WorkflowFactory. Pass "
|
||||
"http_request_handler=DefaultHttpRequestHandler() (or a custom "
|
||||
"implementation) to enable HTTP requests."
|
||||
)
|
||||
executor = HttpRequestActionExecutor(
|
||||
action_def,
|
||||
id=action_id,
|
||||
http_request_handler=self._http_request_handler,
|
||||
)
|
||||
else:
|
||||
executor = executor_class(action_def, id=action_id)
|
||||
self._executors[action_id] = executor
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Error types for declarative workflow executor modules.
|
||||
|
||||
This module exists so that executor modules and the builder (e.g.
|
||||
``_executors_http``, ``_declarative_builder``) can raise declarative-specific
|
||||
exceptions without importing from ``_factory``. ``_factory`` imports
|
||||
``_declarative_builder`` which imports the executor modules; pulling
|
||||
:class:`DeclarativeWorkflowError` from ``_factory`` into an executor or
|
||||
builder module would therefore introduce a circular import.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from agent_framework.exceptions import WorkflowException
|
||||
|
||||
|
||||
class DeclarativeWorkflowError(WorkflowException):
|
||||
"""Raised for build-time / factory-level declarative workflow errors.
|
||||
|
||||
Used for YAML parsing/validation issues, missing configuration (e.g. an
|
||||
HTTP request handler not supplied for a workflow that contains an
|
||||
``HttpRequestAction``), and other errors detected before workflow
|
||||
execution begins.
|
||||
"""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class DeclarativeActionError(WorkflowException):
|
||||
"""Raised when a declarative action fails at run time.
|
||||
|
||||
Used by executor modules for runtime failures (e.g. transport errors,
|
||||
non-2xx responses from :class:`HttpRequestActionExecutor`). Build-time and
|
||||
factory-level errors continue to use :class:`DeclarativeWorkflowError`.
|
||||
"""
|
||||
|
||||
pass
|
||||
+417
@@ -0,0 +1,417 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""Executor for the ``HttpRequestAction`` declarative action.
|
||||
|
||||
Mirrors the .NET ``HttpRequestExecutor``: dispatches an HTTP request through the
|
||||
configured :class:`HttpRequestHandler`, parses the response body, and assigns
|
||||
the parsed body and response headers to the declared state paths.
|
||||
|
||||
Security note: response bodies can echo secrets and may be very large. Diagnostic
|
||||
messages produced for non-2xx responses truncate the body to 256 characters and
|
||||
collapse CR/LF/TAB to spaces (parity with .NET ``FormatBodyForDiagnostics``).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from agent_framework import (
|
||||
Message,
|
||||
WorkflowContext,
|
||||
handler,
|
||||
)
|
||||
|
||||
from ._declarative_base import (
|
||||
ActionComplete,
|
||||
DeclarativeActionExecutor,
|
||||
DeclarativeWorkflowState,
|
||||
)
|
||||
from ._errors import DeclarativeActionError
|
||||
from ._http_handler import HttpRequestHandler, HttpRequestInfo, HttpRequestResult
|
||||
|
||||
__all__ = [
|
||||
"HTTP_ACTION_EXECUTORS",
|
||||
"HttpRequestActionExecutor",
|
||||
]
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_MAX_BODY_DIAGNOSTIC_LENGTH = 256
|
||||
_BODY_TRUNCATION_SUFFIX = " \u2026 [truncated]"
|
||||
|
||||
|
||||
# Body discriminator aliases. Long forms match the .NET object-model type
|
||||
# names so YAML produced by .NET round-trips. Short forms are the .NET YAML
|
||||
# convention used in test fixtures.
|
||||
_BODY_KIND_JSON = {"json", "JsonRequestContent"}
|
||||
_BODY_KIND_RAW = {"raw", "RawRequestContent"}
|
||||
_BODY_KIND_NONE = {"none", "NoRequestContent"}
|
||||
|
||||
|
||||
def _get_path(action_def: Mapping[str, Any], key: str) -> str | None:
|
||||
"""Extract a state path from ``response``/``responseHeaders`` field.
|
||||
|
||||
Supports two YAML shapes (matches .NET serialization round-trips):
|
||||
|
||||
- ``response: Local.MyVar`` (plain string).
|
||||
- ``response: { path: Local.MyVar }`` (object form).
|
||||
"""
|
||||
value = action_def.get(key)
|
||||
if isinstance(value, str):
|
||||
return value or None
|
||||
if isinstance(value, Mapping):
|
||||
path = value.get("path") # type: ignore[reportUnknownMemberType, reportUnknownVariableType]
|
||||
return path if isinstance(path, str) and path else None
|
||||
return None
|
||||
|
||||
|
||||
def _format_body_for_diagnostics(body: str | None) -> str:
|
||||
"""Truncate and sanitise a response body for inclusion in error messages.
|
||||
|
||||
Mirrors the .NET ``FormatBodyForDiagnostics`` helper:
|
||||
|
||||
- Empty/None -> empty string.
|
||||
- Replaces CR/LF/TAB with spaces.
|
||||
- Truncates to 256 chars with a unicode-ellipsis ``[truncated]`` suffix.
|
||||
"""
|
||||
if not body:
|
||||
return ""
|
||||
|
||||
truncated = len(body) > _MAX_BODY_DIAGNOSTIC_LENGTH
|
||||
head = body[:_MAX_BODY_DIAGNOSTIC_LENGTH] if truncated else body
|
||||
sanitized = head.replace("\r", " ").replace("\n", " ").replace("\t", " ")
|
||||
return sanitized + _BODY_TRUNCATION_SUFFIX if truncated else sanitized
|
||||
|
||||
|
||||
def _parse_response_body(body: str | None) -> Any:
|
||||
"""Parse an HTTP response body the same way the .NET executor does.
|
||||
|
||||
JSON-first: if the body parses as JSON, the parsed value is returned. Other
|
||||
bodies are returned as the raw string. Empty/None bodies return ``None``.
|
||||
"""
|
||||
if body is None or body == "":
|
||||
return None
|
||||
try:
|
||||
return json.loads(body)
|
||||
except json.JSONDecodeError:
|
||||
return body
|
||||
|
||||
|
||||
def _format_query_value(value: Any) -> str | None:
|
||||
"""Format a query-parameter value for URL inclusion.
|
||||
|
||||
Mirrors .NET ``FormatQueryValue``: ``None`` is dropped, ``bool`` becomes
|
||||
lower-case ``"true"``/``"false"``, numerics use invariant ``str()``, and
|
||||
other values fall through to ``str()``.
|
||||
"""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
return "true" if value else "false"
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
return str(value)
|
||||
|
||||
|
||||
def _get_messages_path(state: DeclarativeWorkflowState, conversation_id_expr: str | None) -> str | None:
|
||||
"""Return the configured conversation messages path, if any.
|
||||
|
||||
Returns ``System.conversations.{evaluated_id}.messages`` when a
|
||||
``conversation_id_expr`` is configured and evaluates to a non-empty value.
|
||||
Returns ``None`` when no conversation id expression is configured or when
|
||||
the expression evaluates to ``None`` or an empty string (matches .NET
|
||||
``GetConversationId`` behaviour where empty becomes ``null`` and the
|
||||
response is not appended).
|
||||
"""
|
||||
if not conversation_id_expr:
|
||||
return None
|
||||
evaluated = state.eval_if_expression(conversation_id_expr)
|
||||
if evaluated is None or (isinstance(evaluated, str) and not evaluated):
|
||||
return None
|
||||
return f"System.conversations.{evaluated}.messages"
|
||||
|
||||
|
||||
class HttpRequestActionExecutor(DeclarativeActionExecutor):
|
||||
"""Executor for the ``HttpRequestAction`` declarative action.
|
||||
|
||||
Dispatches through the supplied :class:`HttpRequestHandler` and:
|
||||
|
||||
- Parses the response body (JSON-first, raw string fall-back).
|
||||
- Assigns the parsed body to ``response`` path (if configured).
|
||||
- Folds multi-value response headers (comma-joined) and assigns them to
|
||||
``responseHeaders`` path (if configured).
|
||||
- On 2xx with non-empty body and a configured ``conversationId``, appends
|
||||
an Assistant :class:`agent_framework.Message` to
|
||||
``System.conversations.{id}.messages``.
|
||||
- On non-2xx, still publishes ``responseHeaders`` (diagnostic) and raises
|
||||
:class:`DeclarativeActionError` with a status-coded message containing a
|
||||
truncated/sanitised body preview.
|
||||
|
||||
Transport errors (``httpx.TimeoutException``, ``TimeoutError``,
|
||||
``httpx.HTTPError``) become :class:`DeclarativeActionError`. ``CancelledError``
|
||||
is intentionally NOT caught so that workflow cancellation propagates.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
action_def: dict[str, Any],
|
||||
*,
|
||||
id: str | None = None,
|
||||
http_request_handler: HttpRequestHandler,
|
||||
) -> None:
|
||||
"""Create an HTTP request action executor.
|
||||
|
||||
Args:
|
||||
action_def: Parsed ``HttpRequestAction`` YAML dict.
|
||||
id: Optional executor id (defaults to action id or generated).
|
||||
http_request_handler: Handler used to dispatch HTTP requests.
|
||||
Required: the builder enforces presence at workflow-build time.
|
||||
"""
|
||||
super().__init__(action_def, id=id)
|
||||
self._http_request_handler = http_request_handler
|
||||
|
||||
@handler
|
||||
async def handle_action(
|
||||
self,
|
||||
trigger: Any,
|
||||
ctx: WorkflowContext[ActionComplete],
|
||||
) -> None:
|
||||
"""Execute the HTTP request action."""
|
||||
state = await self._ensure_state_initialized(ctx, trigger)
|
||||
|
||||
method = self._get_method(state)
|
||||
url = self._get_url(state)
|
||||
headers = self._get_headers(state)
|
||||
query_parameters = self._get_query_parameters(state)
|
||||
body, body_content_type = self._get_body(state)
|
||||
timeout_ms = self._get_timeout_ms(state)
|
||||
conversation_id_expr = self._action_def.get("conversationId")
|
||||
connection_name = self._get_connection_name(state)
|
||||
|
||||
info = HttpRequestInfo(
|
||||
method=method,
|
||||
url=url,
|
||||
headers=headers or {},
|
||||
query_parameters=query_parameters or {},
|
||||
body=body,
|
||||
body_content_type=body_content_type,
|
||||
timeout_ms=timeout_ms,
|
||||
connection_name=connection_name,
|
||||
)
|
||||
|
||||
try:
|
||||
result = await self._http_request_handler.send(info)
|
||||
except (httpx.TimeoutException, TimeoutError) as exc:
|
||||
raise DeclarativeActionError(f"HTTP request to '{url}' timed out.") from exc
|
||||
except DeclarativeActionError:
|
||||
raise
|
||||
except httpx.HTTPError as exc:
|
||||
raise DeclarativeActionError(f"HTTP request to '{url}' failed: {type(exc).__name__}") from exc
|
||||
except Exception as exc:
|
||||
# Custom HttpRequestHandler implementations may raise arbitrary
|
||||
# exception types. Wrap them in DeclarativeActionError so workflow
|
||||
# error handling stays uniform regardless of transport. Note that
|
||||
# ``asyncio.CancelledError`` is a ``BaseException`` (not
|
||||
# ``Exception``) and so still propagates unmodified, preserving
|
||||
# workflow-cancellation semantics.
|
||||
raise DeclarativeActionError(f"HTTP request to '{url}' failed: {type(exc).__name__}") from exc
|
||||
|
||||
if result.is_success_status_code:
|
||||
self._assign_response(state, result)
|
||||
self._assign_response_headers(state, result)
|
||||
self._append_response_to_conversation(state, conversation_id_expr, result.body)
|
||||
await ctx.send_message(ActionComplete())
|
||||
return
|
||||
|
||||
# Non-success path: still publish headers diagnostically, then raise.
|
||||
self._assign_response_headers(state, result)
|
||||
body_preview = _format_body_for_diagnostics(result.body)
|
||||
if body_preview:
|
||||
message = f"HTTP request to '{url}' failed with status code {result.status_code}. Body: '{body_preview}'"
|
||||
else:
|
||||
message = f"HTTP request to '{url}' failed with status code {result.status_code}."
|
||||
raise DeclarativeActionError(message)
|
||||
|
||||
# ----- Field resolution ----------------------------------------------------
|
||||
|
||||
def _get_method(self, state: DeclarativeWorkflowState) -> str:
|
||||
method = self._action_def.get("method")
|
||||
evaluated = state.eval_if_expression(method) if method is not None else None
|
||||
if not evaluated:
|
||||
return "GET"
|
||||
return str(evaluated).upper()
|
||||
|
||||
def _get_url(self, state: DeclarativeWorkflowState) -> str:
|
||||
raw = self._action_def.get("url")
|
||||
if raw is None:
|
||||
raise ValueError("HttpRequestAction requires a 'url' field.")
|
||||
evaluated = state.eval_if_expression(raw)
|
||||
if not isinstance(evaluated, str) or not evaluated:
|
||||
raise ValueError("HttpRequestAction 'url' evaluated to an empty value.")
|
||||
return evaluated
|
||||
|
||||
def _get_headers(self, state: DeclarativeWorkflowState) -> dict[str, str] | None:
|
||||
raw_headers = self._action_def.get("headers")
|
||||
if not isinstance(raw_headers, Mapping) or not raw_headers:
|
||||
return None
|
||||
result: dict[str, str] = {}
|
||||
for key, value in raw_headers.items(): # type: ignore[reportUnknownVariableType]
|
||||
if not isinstance(key, str) or not key:
|
||||
continue
|
||||
evaluated = state.eval_if_expression(value)
|
||||
if evaluated is None:
|
||||
continue
|
||||
text = str(evaluated)
|
||||
if not text:
|
||||
continue
|
||||
result[key] = text
|
||||
return result or None
|
||||
|
||||
def _get_query_parameters(self, state: DeclarativeWorkflowState) -> dict[str, str] | None:
|
||||
raw_params = self._action_def.get("queryParameters")
|
||||
if not isinstance(raw_params, Mapping) or not raw_params:
|
||||
return None
|
||||
result: dict[str, str] = {}
|
||||
for key, value in raw_params.items(): # type: ignore[reportUnknownVariableType]
|
||||
if not isinstance(key, str) or not key or value is None:
|
||||
continue
|
||||
evaluated = state.eval_if_expression(value)
|
||||
formatted = _format_query_value(evaluated)
|
||||
if formatted is not None:
|
||||
result[key] = formatted
|
||||
return result or None
|
||||
|
||||
def _get_body(self, state: DeclarativeWorkflowState) -> tuple[str | None, str | None]:
|
||||
raw_body = self._action_def.get("body")
|
||||
if raw_body is None:
|
||||
return None, None
|
||||
if not isinstance(raw_body, Mapping):
|
||||
raise ValueError(
|
||||
"HttpRequestAction 'body' must be a mapping with a 'kind' field (json, raw) or omitted entirely."
|
||||
)
|
||||
|
||||
kind_value: Any = raw_body.get("kind") or raw_body.get("$kind") # type: ignore[reportUnknownMemberType]
|
||||
if kind_value is None:
|
||||
raise ValueError(
|
||||
"HttpRequestAction 'body' is missing 'kind'. Use 'json', 'raw', or omit 'body' for no request body."
|
||||
)
|
||||
if not isinstance(kind_value, str):
|
||||
raise ValueError(f"HttpRequestAction 'body.kind' must be a string, got {kind_value!r}.")
|
||||
|
||||
if kind_value in _BODY_KIND_NONE:
|
||||
return None, None
|
||||
|
||||
if kind_value in _BODY_KIND_JSON:
|
||||
content_expr: Any = raw_body.get("content") # type: ignore[reportUnknownMemberType]
|
||||
if content_expr is None:
|
||||
return None, None
|
||||
evaluated = state.eval_if_expression(content_expr)
|
||||
try:
|
||||
body_text = json.dumps(evaluated, default=str)
|
||||
except (TypeError, ValueError) as exc:
|
||||
raise ValueError(f"HttpRequestAction 'body.content' could not be serialised as JSON: {exc}") from exc
|
||||
return body_text, "application/json"
|
||||
|
||||
if kind_value in _BODY_KIND_RAW:
|
||||
content_expr = raw_body.get("content") # type: ignore[reportUnknownMemberType]
|
||||
content_type_expr: Any = raw_body.get("contentType") # type: ignore[reportUnknownMemberType]
|
||||
content: str | None = None
|
||||
if content_expr is not None:
|
||||
evaluated = state.eval_if_expression(content_expr)
|
||||
content = None if evaluated is None else str(evaluated)
|
||||
content_type: str | None = None
|
||||
if content_type_expr is not None:
|
||||
ct_eval = state.eval_if_expression(content_type_expr)
|
||||
ct_text = None if ct_eval is None else str(ct_eval)
|
||||
content_type = ct_text or None
|
||||
# Match .NET RawRequestContent semantics: when a raw body is sent
|
||||
# without an explicit content type, default to text/plain so the
|
||||
# request is interpretable by servers.
|
||||
if content is not None and not content_type:
|
||||
content_type = "text/plain"
|
||||
return content, content_type
|
||||
|
||||
raise ValueError(
|
||||
f"HttpRequestAction 'body.kind' has unsupported value '{kind_value}'. "
|
||||
"Expected one of: json, raw, JsonRequestContent, RawRequestContent, "
|
||||
"NoRequestContent."
|
||||
)
|
||||
|
||||
def _get_timeout_ms(self, state: DeclarativeWorkflowState) -> int | None:
|
||||
raw = self._action_def.get("requestTimeoutInMilliseconds")
|
||||
if raw is None:
|
||||
return None
|
||||
evaluated = state.eval_if_expression(raw)
|
||||
if evaluated is None:
|
||||
return None
|
||||
try:
|
||||
value = int(evaluated)
|
||||
except (TypeError, ValueError):
|
||||
logger.debug(
|
||||
"HttpRequestAction: ignoring non-numeric requestTimeoutInMilliseconds=%r",
|
||||
evaluated,
|
||||
)
|
||||
return None
|
||||
return value if value > 0 else None
|
||||
|
||||
def _get_connection_name(self, state: DeclarativeWorkflowState) -> str | None:
|
||||
connection = self._action_def.get("connection")
|
||||
if not isinstance(connection, Mapping):
|
||||
return None
|
||||
name_expr: Any = connection.get("name") # type: ignore[reportUnknownMemberType]
|
||||
if name_expr is None:
|
||||
return None
|
||||
evaluated = state.eval_if_expression(name_expr)
|
||||
if evaluated is None:
|
||||
return None
|
||||
text = str(evaluated)
|
||||
return text or None
|
||||
|
||||
# ----- Result handling -----------------------------------------------------
|
||||
|
||||
def _assign_response(self, state: DeclarativeWorkflowState, result: HttpRequestResult) -> None:
|
||||
path = _get_path(self._action_def, "response")
|
||||
if path is None:
|
||||
return
|
||||
state.set(path, _parse_response_body(result.body))
|
||||
|
||||
def _assign_response_headers(self, state: DeclarativeWorkflowState, result: HttpRequestResult) -> None:
|
||||
path = _get_path(self._action_def, "responseHeaders")
|
||||
if path is None:
|
||||
return
|
||||
if not result.headers:
|
||||
state.set(path, None)
|
||||
return
|
||||
# Fold multi-value headers with commas (standard HTTP folding) only at
|
||||
# assignment time. The raw multi-value dict on HttpRequestResult.headers
|
||||
# is left untouched so callers/tests can inspect duplicates.
|
||||
flattened: dict[str, str] = {}
|
||||
for key, values in result.headers.items():
|
||||
flattened[key] = ",".join(values)
|
||||
state.set(path, flattened)
|
||||
|
||||
def _append_response_to_conversation(
|
||||
self,
|
||||
state: DeclarativeWorkflowState,
|
||||
conversation_id_expr: str | None,
|
||||
body: str,
|
||||
) -> None:
|
||||
if not body:
|
||||
return
|
||||
messages_path = _get_messages_path(state, conversation_id_expr)
|
||||
if messages_path is None:
|
||||
return
|
||||
# Mirrors InvokeAzureAgentExecutor: rely on state.append to lazily
|
||||
# create the conversation entry. Avoids re-parsing the id back out
|
||||
# of the dotted path string.
|
||||
message = Message(role="assistant", contents=[body])
|
||||
state.append(messages_path, message)
|
||||
|
||||
|
||||
HTTP_ACTION_EXECUTORS: dict[str, type[DeclarativeActionExecutor]] = {
|
||||
"HttpRequestAction": HttpRequestActionExecutor,
|
||||
}
|
||||
@@ -24,18 +24,16 @@ from agent_framework import (
|
||||
SupportsAgentRun,
|
||||
Workflow,
|
||||
)
|
||||
from agent_framework.exceptions import WorkflowException
|
||||
|
||||
from .._loader import AgentFactory
|
||||
from ._declarative_builder import DeclarativeWorkflowBuilder
|
||||
from ._errors import DeclarativeWorkflowError
|
||||
from ._http_handler import HttpRequestHandler
|
||||
|
||||
logger = logging.getLogger("agent_framework.declarative")
|
||||
|
||||
|
||||
class DeclarativeWorkflowError(WorkflowException):
|
||||
"""Exception raised for errors in declarative workflow processing."""
|
||||
|
||||
pass
|
||||
__all__ = ["WorkflowFactory"]
|
||||
|
||||
|
||||
class WorkflowFactory:
|
||||
@@ -92,6 +90,7 @@ class WorkflowFactory:
|
||||
env_file: str | None = None,
|
||||
checkpoint_storage: CheckpointStorage | None = None,
|
||||
max_iterations: int | None = None,
|
||||
http_request_handler: HttpRequestHandler | None = None,
|
||||
) -> None:
|
||||
"""Initialize the workflow factory.
|
||||
|
||||
@@ -105,6 +104,12 @@ class WorkflowFactory:
|
||||
max_iterations: Optional maximum runner supersteps. Overrides the YAML ``maxTurns``
|
||||
field and the core default (100). Workflows with ``GotoAction`` loops (e.g.
|
||||
DeepResearch) typically need a higher value.
|
||||
http_request_handler: Optional handler used to dispatch HTTP requests for
|
||||
``HttpRequestAction``. Required if the workflow contains any
|
||||
``HttpRequestAction``; build will fail with :class:`DeclarativeWorkflowError`
|
||||
otherwise. Use :class:`agent_framework.declarative.DefaultHttpRequestHandler`
|
||||
for a no-policy ``httpx``-based default, or supply your own implementation
|
||||
to enforce SSRF guards, allowlisting, or auth resolution.
|
||||
|
||||
Examples:
|
||||
.. code-block:: python
|
||||
@@ -144,6 +149,7 @@ class WorkflowFactory:
|
||||
self._tools: dict[str, Any] = {} # Tool registry for InvokeFunctionTool actions
|
||||
self._checkpoint_storage = checkpoint_storage
|
||||
self._max_iterations = max_iterations
|
||||
self._http_request_handler = http_request_handler
|
||||
|
||||
def create_workflow_from_yaml_path(
|
||||
self,
|
||||
@@ -387,6 +393,7 @@ class WorkflowFactory:
|
||||
tools=self._tools,
|
||||
checkpoint_storage=self._checkpoint_storage,
|
||||
max_iterations=self._max_iterations,
|
||||
http_request_handler=self._http_request_handler,
|
||||
)
|
||||
workflow = graph_builder.build()
|
||||
except ValueError as e:
|
||||
|
||||
@@ -0,0 +1,237 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
"""HTTP request handler abstraction for declarative workflows.
|
||||
|
||||
Mirrors the .NET ``IHttpRequestHandler`` / ``DefaultHttpRequestHandler`` pair from
|
||||
``Microsoft.Agents.AI.Workflows.Declarative``. Provides:
|
||||
|
||||
- :class:`HttpRequestInfo` — request input data passed from the executor.
|
||||
- :class:`HttpRequestResult` — response data returned to the executor.
|
||||
- :class:`HttpRequestHandler` — :class:`typing.Protocol` callers implement to plug
|
||||
in custom transports (e.g. with allowlisting, mTLS, retries, etc.).
|
||||
- :class:`DefaultHttpRequestHandler` — production-grade default backed by
|
||||
``httpx.AsyncClient``.
|
||||
|
||||
Security note: :class:`DefaultHttpRequestHandler` performs **no** URL filtering
|
||||
or SSRF protection. Production deployments should supply a custom handler that
|
||||
enforces an allowlist or DNS-rebinding-resistant policy. This split mirrors the
|
||||
.NET design.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from collections.abc import Awaitable, Callable, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Protocol, runtime_checkable
|
||||
|
||||
import httpx
|
||||
|
||||
__all__ = [
|
||||
"DefaultHttpRequestHandler",
|
||||
"HttpRequestHandler",
|
||||
"HttpRequestInfo",
|
||||
"HttpRequestResult",
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class HttpRequestInfo:
|
||||
"""Description of an HTTP request to be dispatched by a :class:`HttpRequestHandler`.
|
||||
|
||||
Mirrors the .NET ``HttpRequestInfo`` record. Field semantics:
|
||||
|
||||
- ``method``: HTTP method (``GET``, ``POST``, etc.). Already upper-cased by the executor.
|
||||
- ``url``: Absolute URL. Already evaluated from the YAML expression.
|
||||
- ``headers``: Single-value header map (case-insensitive keys per HTTP semantics
|
||||
but stored as authored). Empty values are skipped by the executor.
|
||||
- ``query_parameters``: String key/value pairs appended to the URL.
|
||||
- ``body``: Request body bytes/text, or ``None`` for no body.
|
||||
- ``body_content_type``: Content type to send (e.g. ``application/json``).
|
||||
Ignored when ``body`` is ``None``.
|
||||
- ``timeout_ms``: Per-request timeout in milliseconds. ``None`` => use the
|
||||
handler's default.
|
||||
- ``connection_name``: Optional Foundry connection name for handlers that
|
||||
resolve auth/credentials by connection.
|
||||
"""
|
||||
|
||||
method: str
|
||||
url: str
|
||||
headers: dict[str, str] = field(default_factory=dict) # type: ignore[reportUnknownVariableType]
|
||||
query_parameters: dict[str, str] = field(default_factory=dict) # type: ignore[reportUnknownVariableType]
|
||||
body: str | None = None
|
||||
body_content_type: str | None = None
|
||||
timeout_ms: int | None = None
|
||||
connection_name: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class HttpRequestResult:
|
||||
"""Response returned by a :class:`HttpRequestHandler`.
|
||||
|
||||
Mirrors the .NET ``HttpRequestResult`` record. ``headers`` preserves
|
||||
multi-value response headers (e.g. multiple ``Set-Cookie`` headers) as a
|
||||
``dict[str, list[str]]``. The executor folds duplicates into a single
|
||||
comma-joined string only at the point it assigns ``responseHeaders`` to
|
||||
workflow state.
|
||||
|
||||
Header keys are normalized to lowercase so that lookups are consistent
|
||||
regardless of the server's transmitted casing (HTTP headers are
|
||||
case-insensitive per RFC 7230 §3.2). Custom :class:`HttpRequestHandler`
|
||||
implementations should follow the same convention.
|
||||
"""
|
||||
|
||||
status_code: int
|
||||
is_success_status_code: bool
|
||||
body: str
|
||||
headers: dict[str, list[str]] = field(default_factory=dict) # type: ignore[reportUnknownVariableType]
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class HttpRequestHandler(Protocol):
|
||||
"""Protocol for HTTP request handlers used by ``HttpRequestAction``.
|
||||
|
||||
Implementations must be safe to call concurrently from multiple workflow
|
||||
runs. Implementations are responsible for any URL allowlisting, SSRF
|
||||
guards, retry policies, auth resolution, and other policies that the
|
||||
workflow author wants applied.
|
||||
"""
|
||||
|
||||
async def send(self, info: HttpRequestInfo) -> HttpRequestResult:
|
||||
"""Dispatch ``info`` and return the response result.
|
||||
|
||||
Args:
|
||||
info: Description of the request to send.
|
||||
|
||||
Returns:
|
||||
The response. Implementations should NOT raise on non-2xx status
|
||||
codes; instead, set ``is_success_status_code`` accordingly. They
|
||||
SHOULD raise on transport-level failures (connection refused,
|
||||
DNS errors, timeouts).
|
||||
"""
|
||||
...
|
||||
|
||||
|
||||
ClientProvider = Callable[[HttpRequestInfo], Awaitable["httpx.AsyncClient | None"]]
|
||||
|
||||
|
||||
class DefaultHttpRequestHandler:
|
||||
"""Default :class:`HttpRequestHandler` backed by :class:`httpx.AsyncClient`.
|
||||
|
||||
Construction modes:
|
||||
|
||||
1. ``DefaultHttpRequestHandler()`` — owns an internal client created lazily
|
||||
on first ``send()``. Closed by :meth:`aclose`.
|
||||
2. ``DefaultHttpRequestHandler(client=existing)`` — caller-owned client.
|
||||
Not closed by :meth:`aclose`.
|
||||
3. ``DefaultHttpRequestHandler(client_provider=cb)`` — per-request client
|
||||
lookup (parity with .NET's ``httpClientProvider`` callback). The
|
||||
provider may return ``None`` to fall back to the owned/default client.
|
||||
|
||||
.. warning::
|
||||
|
||||
This handler performs **no** URL filtering or SSRF protection. Wrap or
|
||||
replace it with a custom handler in production.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
client: httpx.AsyncClient | None = None,
|
||||
client_provider: ClientProvider | None = None,
|
||||
) -> None:
|
||||
self._owned_client: httpx.AsyncClient | None = None
|
||||
self._caller_client = client
|
||||
self._client_provider = client_provider
|
||||
# Guards lazy creation of ``_owned_client`` against concurrent first
|
||||
# ``send()`` calls leaking duplicate clients.
|
||||
self._owned_client_lock = asyncio.Lock()
|
||||
|
||||
async def send(self, info: HttpRequestInfo) -> HttpRequestResult:
|
||||
"""Dispatch the request and return the parsed result."""
|
||||
if not info.url:
|
||||
raise ValueError("HttpRequestInfo.url must be a non-empty string.")
|
||||
if not info.method:
|
||||
raise ValueError("HttpRequestInfo.method must be a non-empty string.")
|
||||
|
||||
client = await self._resolve_client(info)
|
||||
|
||||
timeout: httpx.Timeout | object
|
||||
if info.timeout_ms is not None and info.timeout_ms > 0:
|
||||
timeout = httpx.Timeout(info.timeout_ms / 1000.0)
|
||||
else:
|
||||
timeout = httpx.USE_CLIENT_DEFAULT
|
||||
|
||||
headers = dict(info.headers)
|
||||
content: bytes | str | None = None
|
||||
if info.body is not None:
|
||||
content = info.body
|
||||
if not _has_header(headers, "content-type"):
|
||||
# Match .NET DefaultHttpRequestHandler: when a body is sent
|
||||
# without an explicit content type, default to ``text/plain``
|
||||
# so the request is interpretable by servers and direct
|
||||
# callers (not just the YAML executor) get sensible defaults.
|
||||
headers["Content-Type"] = info.body_content_type or "text/plain"
|
||||
|
||||
params: Mapping[str, str] | None = info.query_parameters or None
|
||||
|
||||
response = await client.request(
|
||||
method=info.method,
|
||||
url=info.url,
|
||||
params=params,
|
||||
headers=headers or None,
|
||||
content=content,
|
||||
timeout=timeout, # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
# Preserve multi-value headers (e.g. multiple Set-Cookie) as list[str].
|
||||
# Normalize names to lowercase so lookups are consistent and case
|
||||
# variations from the transport do not create duplicate logical keys
|
||||
# (HTTP headers are case-insensitive per RFC 7230 §3.2).
|
||||
result_headers: dict[str, list[str]] = {}
|
||||
for key, value in response.headers.multi_items():
|
||||
result_headers.setdefault(key.lower(), []).append(value)
|
||||
|
||||
body_text = response.text
|
||||
|
||||
return HttpRequestResult(
|
||||
status_code=response.status_code,
|
||||
is_success_status_code=200 <= response.status_code < 300,
|
||||
body=body_text,
|
||||
headers=result_headers,
|
||||
)
|
||||
|
||||
async def aclose(self) -> None:
|
||||
"""Release the owned client, if any. Caller-owned clients are NOT closed."""
|
||||
if self._owned_client is not None:
|
||||
await self._owned_client.aclose()
|
||||
self._owned_client = None
|
||||
|
||||
async def _resolve_client(self, info: HttpRequestInfo) -> httpx.AsyncClient:
|
||||
"""Pick a client for this request: provider → caller → lazily-owned."""
|
||||
if self._client_provider is not None:
|
||||
provided = await self._client_provider(info)
|
||||
if provided is not None:
|
||||
return provided
|
||||
if self._caller_client is not None:
|
||||
return self._caller_client
|
||||
if self._owned_client is None:
|
||||
# Double-checked locking under asyncio.Lock so concurrent first
|
||||
# callers don't each create a fresh httpx.AsyncClient and orphan
|
||||
# one of them.
|
||||
async with self._owned_client_lock:
|
||||
if self._owned_client is None:
|
||||
self._owned_client = httpx.AsyncClient()
|
||||
return self._owned_client
|
||||
|
||||
async def __aenter__(self) -> DefaultHttpRequestHandler:
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
|
||||
await self.aclose()
|
||||
|
||||
|
||||
def _has_header(headers: Mapping[str, str], name: str) -> bool:
|
||||
"""Case-insensitive header presence check."""
|
||||
needle = name.lower()
|
||||
return any(key.lower() == needle for key in headers)
|
||||
Reference in New Issue
Block a user