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: Fix Http Schema (#2112)

Browse Source 
* Rename to threadid

* Respond in plain text

* Make snake-case

* Add http prefix

* rename to wait-for-response

* Add query param check

* address comments
This commit is contained in:
Laveesh Rohra
2025-11-12 09:56:19 -08:00
committed by GitHub
Unverified
parent ebab25b196
commit ff28066c9c
24 changed files with 692 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/azurefunctions/agent_framework_azurefunctions/_app.py
+200 -118
View File
@@ -8,7 +8,7 @@ with Azure Durable Entities, enabling stateful and durable AI agent execution.
import json
import re
from collections.abc import Mapping
from collections.abc import Callable, Mapping
from typing import Any, cast
import azure.durable_functions as df
@@ -18,17 +18,16 @@ from agent_framework import AgentProtocol, get_logger
from ._callbacks import AgentResponseCallbackProtocol
from ._entities import create_agent_entity
from ._errors import IncomingRequestError
from ._models import AgentSessionId, ChatRole, RunRequest
from ._models import AgentSessionId, RunRequest
from ._state import AgentState
logger = get_logger("agent_framework.azurefunctions")
SESSION_ID_FIELD: str = "sessionId"
SESSION_KEY_FIELD: str = "sessionKey"
SESSION_IDENTIFIER_KEYS: tuple[str, str] = (
SESSION_ID_FIELD,
SESSION_KEY_FIELD,
)
THREAD_ID_FIELD: str = "thread_id"
RESPONSE_FORMAT_JSON: str = "json"
RESPONSE_FORMAT_TEXT: str = "text"
WAIT_FOR_RESPONSE_FIELD: str = "wait_for_response"
WAIT_FOR_RESPONSE_HEADER: str = "x-ms-wait-for-response"
class AgentFunctionApp(df.DFApp):
@@ -88,18 +87,18 @@ class AgentFunctionApp(df.DFApp):
def __init__(
self,
agents: list[AgentProtocol] | None = None,
http_auth_level: func.AuthLevel = func.AuthLevel.ANONYMOUS,
http_auth_level: func.AuthLevel = func.AuthLevel.FUNCTION,
enable_health_check: bool = True,
enable_http_endpoints: bool = True,
max_poll_retries: int = 10,
poll_interval_seconds: float = 0.5,
max_poll_retries: int = 30,
poll_interval_seconds: float = 1,
default_callback: AgentResponseCallbackProtocol | None = None,
):
"""Initialize the AgentFunctionApp.
Args:
agents: List of agent instances to register
http_auth_level: HTTP authentication level (default: ANONYMOUS)
http_auth_level: HTTP authentication level (default: FUNCTION)
enable_health_check: Enable built-in health check endpoint (default: True)
enable_http_endpoints: Enable HTTP endpoints for agents (default: True)
max_poll_retries: Maximum number of polling attempts when waiting for a response
@@ -231,7 +230,7 @@ class AgentFunctionApp(df.DFApp):
Args:
agent_name: The agent name (used for both routing and entity identification)
"""
run_function_name = self._build_function_name(agent_name, "run")
run_function_name = self._build_function_name(agent_name, "http")
@self.function_name(run_function_name)
@self.route(route=f"agents/{agent_name}/run", methods=["POST"])
@@ -242,7 +241,7 @@ class AgentFunctionApp(df.DFApp):
Expected request body (RunRequest format):
{
"message": "user message to agent",
"sessionId": "optional session id (or sessionKey)",
"thread_id": "optional conversation identifier",
"role": "user|system" (optional, default: "user"),
"response_format": {...} (optional JSON schema for structured responses),
"enable_tool_calls": true|false (optional, default: true)
@@ -250,22 +249,28 @@ class AgentFunctionApp(df.DFApp):
"""
logger.debug(f"[HTTP Trigger] Received request on route: /api/agents/{agent_name}/run")
response_format: str = RESPONSE_FORMAT_JSON
thread_id: str | None = None
try:
req_body, message = self._parse_incoming_request(req)
session_key = self._resolve_session_key(req=req, req_body=req_body)
wait_for_completion = self._should_wait_for_completion(req=req, req_body=req_body)
req_body, message, response_format = self._parse_incoming_request(req)
thread_id = self._resolve_thread_id(req=req, req_body=req_body)
wait_for_response = self._should_wait_for_response(req=req, req_body=req_body)
logger.debug(f"[HTTP Trigger] Message: {message}")
logger.debug(f"[HTTP Trigger] Session Key: {session_key}")
logger.debug(f"[HTTP Trigger] wait_for_completion: {wait_for_completion}")
logger.debug(f"[HTTP Trigger] Thread ID: {thread_id}")
logger.debug(f"[HTTP Trigger] wait_for_response: {wait_for_response}")
if not message:
logger.warning("[HTTP Trigger] Request rejected: Missing message")
return func.HttpResponse(
json.dumps({"error": "Message is required"}), status_code=400, mimetype="application/json"
return self._create_http_response(
payload={"error": "Message is required"},
status_code=400,
response_format=response_format,
thread_id=thread_id,
)
session_id = self._create_session_id(agent_name, session_key)
session_id = self._create_session_id(agent_name, thread_id)
correlation_id = self._generate_unique_id()
logger.debug(f"[HTTP Trigger] Using session ID: {session_id}")
@@ -276,7 +281,7 @@ class AgentFunctionApp(df.DFApp):
run_request = self._build_request_data(
req_body,
message,
session_key,
thread_id,
correlation_id,
)
logger.debug("Signalling entity %s with request: %s", entity_instance_id, run_request)
@@ -284,43 +289,60 @@ class AgentFunctionApp(df.DFApp):
logger.debug(f"[HTTP Trigger] Signal sent to entity {session_id}")
if wait_for_completion:
if wait_for_response:
result = await self._get_response_from_entity(
client=client,
entity_instance_id=entity_instance_id,
correlation_id=correlation_id,
message=message,
session_key=session_key,
thread_id=thread_id,
)
logger.debug(f"[HTTP Trigger] Result status: {result.get('status', 'unknown')}")
return func.HttpResponse(
json.dumps(result),
return self._create_http_response(
payload=result,
status_code=200 if result.get("status") == "success" else 500,
mimetype="application/json",
response_format=response_format,
thread_id=thread_id,
)
logger.debug("[HTTP Trigger] wait_for_completion disabled; returning correlation ID")
logger.debug("[HTTP Trigger] wait_for_response disabled; returning correlation ID")
accepted_response = self._build_accepted_response(
message=message, session_key=session_key, correlation_id=correlation_id
message=message, thread_id=thread_id, correlation_id=correlation_id
)
return func.HttpResponse(json.dumps(accepted_response), status_code=202, mimetype="application/json")
return self._create_http_response(
payload=accepted_response,
status_code=202,
response_format=response_format,
thread_id=thread_id,
)
except IncomingRequestError as exc:
logger.warning(f"[HTTP Trigger] Request rejected: {exc!s}")
return func.HttpResponse(
json.dumps({"error": str(exc)}), status_code=exc.status_code, mimetype="application/json"
return self._create_http_response(
payload={"error": str(exc)},
status_code=exc.status_code,
response_format=response_format,
thread_id=thread_id,
)
except ValueError as exc:
logger.error(f"[HTTP Trigger] Invalid JSON: {exc!s}")
return func.HttpResponse(
json.dumps({"error": "Invalid JSON"}), status_code=400, mimetype="application/json"
return self._create_http_response(
payload={"error": "Invalid JSON"},
status_code=400,
response_format=response_format,
thread_id=thread_id,
)
except Exception as exc:
logger.error(f"[HTTP Trigger] Error: {exc!s}", exc_info=True)
return func.HttpResponse(json.dumps({"error": str(exc)}), status_code=500, mimetype="application/json")
return self._create_http_response(
payload={"error": str(exc)},
status_code=500,
response_format=response_format,
thread_id=thread_id,
)
_ = http_start
@@ -365,7 +387,7 @@ class AgentFunctionApp(df.DFApp):
{
"name": name,
"type": type(agent).__name__,
"httpEndpointEnabled": self.agent_http_endpoint_flags.get(
"http_endpoint_enabled": self.agent_http_endpoint_flags.get(
name,
self.enable_http_endpoints,
),
@@ -381,17 +403,20 @@ class AgentFunctionApp(df.DFApp):
_ = health_check
@staticmethod
def _build_function_name(agent_name: str, suffix: str) -> str:
"""Generate a unique, Azure Functions-compliant name for an agent function."""
sanitized = re.sub(r"[^0-9a-zA-Z_]", "_", agent_name or "agent").strip("_")
def _build_function_name(agent_name: str, prefix: str) -> str:
"""Generate the sanitized function name in the form "{prefix}-{sanitized_agent_name}".
if not sanitized:
sanitized = "agent"
Example: agent_name="Weather Agent" and prefix="http" becomes "http-Weather_Agent".
"""
sanitized_agent = re.sub(r"[^0-9a-zA-Z_]", "_", agent_name or "agent").strip("_")
if sanitized[0].isdigit():
sanitized = f"agent_{sanitized}"
if not sanitized_agent:
sanitized_agent = "agent"
return f"{sanitized}_{suffix}"
if sanitized_agent[0].isdigit():
sanitized_agent = f"agent_{sanitized_agent}"
return f"{prefix}-{sanitized_agent}"
async def _read_cached_state(
self,
@@ -418,7 +443,7 @@ class AgentFunctionApp(df.DFApp):
entity_instance_id: df.EntityId,
correlation_id: str,
message: str,
session_key: str,
thread_id: str,
) -> dict[str, Any]:
"""Poll the entity state until a response is available or timeout occurs."""
import asyncio
@@ -438,7 +463,7 @@ class AgentFunctionApp(df.DFApp):
entity_instance_id=entity_instance_id,
correlation_id=correlation_id,
message=message,
session_key=session_key,
thread_id=thread_id,
)
if result is not None:
break
@@ -453,7 +478,7 @@ class AgentFunctionApp(df.DFApp):
f"[HTTP Trigger] Response with correlation ID {correlation_id} "
f"not found in time (waited {max_retries * interval} seconds)"
)
return await self._build_timeout_result(message=message, session_key=session_key, correlation_id=correlation_id)
return await self._build_timeout_result(message=message, thread_id=thread_id, correlation_id=correlation_id)
async def _poll_entity_for_response(
self,
@@ -461,7 +486,7 @@ class AgentFunctionApp(df.DFApp):
entity_instance_id: df.EntityId,
correlation_id: str,
message: str,
session_key: str,
thread_id: str,
) -> dict[str, Any] | None:
result: dict[str, Any] | None = None
try:
@@ -475,7 +500,7 @@ class AgentFunctionApp(df.DFApp):
result = self._build_success_result(
response_data=agent_response,
message=message,
session_key=session_key,
thread_id=thread_id,
correlation_id=correlation_id,
state=state,
)
@@ -486,113 +511,181 @@ class AgentFunctionApp(df.DFApp):
return result
async def _build_timeout_result(self, message: str, session_key: str, correlation_id: str) -> dict[str, Any]:
async def _build_timeout_result(self, message: str, thread_id: str, correlation_id: str) -> dict[str, Any]:
"""Create the timeout response."""
return {
"response": "Agent is still processing or timed out...",
"message": message,
SESSION_ID_FIELD: session_key,
THREAD_ID_FIELD: thread_id,
"status": "timeout",
"correlationId": correlation_id,
"correlation_id": correlation_id,
}
def _build_success_result(
self, response_data: dict[str, Any], message: str, session_key: str, correlation_id: str, state: AgentState
self, response_data: dict[str, Any], message: str, thread_id: str, correlation_id: str, state: AgentState
) -> dict[str, Any]:
"""Build the success result returned to the HTTP caller."""
return {
"response": response_data.get("content"),
"message": message,
SESSION_ID_FIELD: session_key,
THREAD_ID_FIELD: thread_id,
"status": "success",
"message_count": response_data.get("message_count", state.message_count),
"correlationId": correlation_id,
"correlation_id": correlation_id,
}
def _build_request_data(
self, req_body: dict[str, Any], message: str, conversation_id: str, correlation_id: str
self, req_body: dict[str, Any], message: str, thread_id: str, correlation_id: str
) -> dict[str, Any]:
"""Create the durable entity request payload."""
enable_tool_calls_value = req_body.get("enable_tool_calls")
enable_tool_calls = True if enable_tool_calls_value is None else self._coerce_to_bool(enable_tool_calls_value)
role = self._coerce_chat_role(req_body.get("role"))
return RunRequest(
message=message,
role=role,
role=req_body.get("role"),
response_format=req_body.get("response_format"),
enable_tool_calls=enable_tool_calls,
conversation_id=conversation_id,
thread_id=thread_id,
correlation_id=correlation_id,
).to_dict()
def _build_accepted_response(self, message: str, session_key: str, correlation_id: str) -> dict[str, Any]:
def _build_accepted_response(self, message: str, thread_id: str, correlation_id: str) -> dict[str, Any]:
"""Build the response returned when not waiting for completion."""
return {
"response": "Agent request accepted",
"message": message,
SESSION_ID_FIELD: session_key,
THREAD_ID_FIELD: thread_id,
"status": "accepted",
"correlationId": correlation_id,
"correlation_id": correlation_id,
}
def _create_http_response(
self,
payload: dict[str, Any] | str,
status_code: int,
response_format: str,
thread_id: str | None,
) -> func.HttpResponse:
"""Create the HTTP response using helper serializers for clarity."""
if response_format == RESPONSE_FORMAT_TEXT:
return self._build_plain_text_response(payload=payload, status_code=status_code, thread_id=thread_id)
return self._build_json_response(payload=payload, status_code=status_code)
def _build_plain_text_response(
self,
payload: dict[str, Any] | str,
status_code: int,
thread_id: str | None,
) -> func.HttpResponse:
"""Return a plain-text response with optional thread identifier header."""
body_text = payload if isinstance(payload, str) else self._convert_payload_to_text(payload)
headers = {"x-ms-thread-id": thread_id} if thread_id is not None else None
return func.HttpResponse(body_text, status_code=status_code, mimetype="text/plain", headers=headers)
def _build_json_response(self, payload: dict[str, Any] | str, status_code: int) -> func.HttpResponse:
"""Return the JSON response, serializing dictionaries as needed."""
body_json = payload if isinstance(payload, str) else json.dumps(payload)
return func.HttpResponse(body_json, status_code=status_code, mimetype="application/json")
def _convert_payload_to_text(self, payload: dict[str, Any]) -> str:
"""Convert a structured payload into a human-readable text response."""
for key in ("response", "error", "message"):
value = payload.get(key)
if isinstance(value, str) and value:
return value
return json.dumps(payload)
def _generate_unique_id(self) -> str:
"""Generate a new unique identifier."""
import uuid
return uuid.uuid4().hex
def _create_session_id(self, func_name: str, session_key: str | None) -> AgentSessionId:
"""Create a session identifier using the provided key or a random value."""
if session_key:
return AgentSessionId(name=func_name, key=session_key)
def _create_session_id(self, func_name: str, thread_id: str | None) -> AgentSessionId:
"""Create a session identifier using the provided thread id or a random value."""
if thread_id:
return AgentSessionId(name=func_name, key=thread_id)
return AgentSessionId.with_random_key(name=func_name)
def _resolve_session_key(self, req: func.HttpRequest, req_body: dict[str, Any]) -> str:
"""Retrieve the session key from request body or query parameters."""
def _resolve_thread_id(self, req: func.HttpRequest, req_body: dict[str, Any]) -> str:
"""Retrieve the thread identifier from request body or query parameters."""
params = req.params or {}
for key in SESSION_IDENTIFIER_KEYS:
if key in req_body:
value = req_body.get(key)
if value is not None:
return str(value)
if THREAD_ID_FIELD in req_body:
value = req_body.get(THREAD_ID_FIELD)
if value is not None:
return str(value)
for key in SESSION_IDENTIFIER_KEYS:
if key in params:
value = params.get(key)
if value is not None:
return str(value)
if THREAD_ID_FIELD in params:
value = params.get(THREAD_ID_FIELD)
if value is not None:
return str(value)
logger.debug("[HTTP Trigger] No session identifier provided; using random session key")
logger.debug("[HTTP Trigger] No thread identifier provided; using random thread id")
return self._generate_unique_id()
def _parse_incoming_request(self, req: func.HttpRequest) -> tuple[dict[str, Any], Any]:
def _parse_incoming_request(self, req: func.HttpRequest) -> tuple[dict[str, Any], str, str]:
"""Parse the incoming run request supporting JSON and plain text bodies."""
headers = self._extract_normalized_headers(req)
normalized_content_type = self._extract_content_type(headers)
body_parser, body_format = self._select_body_parser(normalized_content_type)
prefers_json = self._accepts_json_response(headers)
response_format = self._select_response_format(body_format=body_format, prefers_json=prefers_json)
req_body, message = body_parser(req)
return req_body, message, response_format
def _extract_normalized_headers(self, req: func.HttpRequest) -> dict[str, str]:
"""Create a lowercase header mapping from the incoming request."""
headers: dict[str, str] = {}
raw_headers = req.headers
if isinstance(raw_headers, Mapping):
headers_mapping = cast(Mapping[Any, Any], raw_headers)
for key, value in headers_mapping.items():
if value is not None:
headers[str(key)] = str(value)
content_type_header = headers.get("content-type")
normalized_content_type = ""
if content_type_header:
normalized_content_type = content_type_header.split(";")[0].strip().lower()
if normalized_content_type in {"application/json"} or normalized_content_type.endswith("+json"):
parser = self._parse_json_body
else:
parser = self._parse_text_body
return parser(req)
headers[str(key).lower()] = str(value)
return headers
@staticmethod
def _parse_json_body(req: func.HttpRequest) -> tuple[dict[str, Any], Any]:
def _extract_content_type(headers: dict[str, str]) -> str:
"""Return the normalized content-type value (without parameters)."""
content_type_header = headers.get("content-type", "")
return content_type_header.split(";")[0].strip().lower() if content_type_header else ""
def _select_body_parser(
self,
normalized_content_type: str,
) -> tuple[Callable[[func.HttpRequest], tuple[dict[str, Any], str]], str]:
"""Choose the body parser and declared body format."""
if normalized_content_type in {"application/json"} or normalized_content_type.endswith("+json"):
return self._parse_json_body, RESPONSE_FORMAT_JSON
return self._parse_text_body, RESPONSE_FORMAT_TEXT
@staticmethod
def _accepts_json_response(headers: dict[str, str]) -> bool:
"""Check whether the caller explicitly requests a JSON response."""
accept_header = headers.get("accept")
if not accept_header:
return False
for value in accept_header.split(","):
media_type = value.split(";")[0].strip().lower()
if media_type == "application/json":
return True
return False
@staticmethod
def _select_response_format(body_format: str, prefers_json: bool) -> str:
"""Combine body format and accept preference to determine response format."""
if body_format == RESPONSE_FORMAT_JSON or prefers_json:
return RESPONSE_FORMAT_JSON
return RESPONSE_FORMAT_TEXT
@staticmethod
def _parse_json_body(req: func.HttpRequest) -> tuple[dict[str, Any], str]:
req_body = req.get_json()
if not isinstance(req_body, dict):
raise IncomingRequestError("Invalid JSON payload. Expected an object.")
@@ -603,46 +696,35 @@ class AgentFunctionApp(df.DFApp):
return typed_req_body, message
@staticmethod
def _parse_text_body(req: func.HttpRequest) -> tuple[dict[str, Any], Any]:
def _parse_text_body(req: func.HttpRequest) -> tuple[dict[str, Any], str]:
body_bytes = req.get_body()
text_body = body_bytes.decode("utf-8", errors="replace") if body_bytes else ""
message = text_body.strip()
if not message:
raise IncomingRequestError("Message is required")
return {}, message
def _should_wait_for_completion(self, req: func.HttpRequest, req_body: dict[str, Any]) -> bool:
"""Determine whether the caller requested to wait for completion."""
def _should_wait_for_response(self, req: func.HttpRequest, req_body: dict[str, Any]) -> bool:
"""Determine whether the caller requested to wait for the response."""
header_value = None
raw_headers = req.headers
if isinstance(raw_headers, Mapping):
headers_mapping = cast(Mapping[Any, Any], raw_headers)
for key, value in headers_mapping.items():
if str(key).lower() == "x-wait-for-completion":
if str(key).lower() == WAIT_FOR_RESPONSE_HEADER:
header_value = value
break
if header_value is not None:
return self._coerce_to_bool(header_value)
for key in ("wait_for_completion", "waitForCompletion", "WaitForCompletion"):
if key in req_body:
return self._coerce_to_bool(req_body.get(key))
params = req.params or {}
if WAIT_FOR_RESPONSE_FIELD in params:
return self._coerce_to_bool(params.get(WAIT_FOR_RESPONSE_FIELD))
return False
if WAIT_FOR_RESPONSE_FIELD in req_body:
return self._coerce_to_bool(req_body.get(WAIT_FOR_RESPONSE_FIELD))
def _coerce_chat_role(self, value: Any) -> ChatRole:
"""Convert user-provided role to ChatRole, defaulting to user on error."""
if isinstance(value, ChatRole):
return value
if isinstance(value, str):
try:
return ChatRole(value.strip().lower())
except ValueError:
logger.warning("[AgentFunctionApp] Invalid role '%s'; defaulting to user", value)
return ChatRole.USER
return True
def _coerce_to_bool(self, value: Any) -> bool:
"""Convert various representations into a boolean flag."""
python/packages/azurefunctions/agent_framework_azurefunctions/_callbacks.py
+1 -1
View File
@@ -20,7 +20,7 @@ class AgentCallbackContext:
agent_name: str
correlation_id: str
conversation_id: str | None = None
thread_id: str | None = None
request_message: str | None = None
python/packages/azurefunctions/agent_framework_azurefunctions/_entities.py
+17 -21
View File
@@ -14,10 +14,10 @@ from collections.abc import AsyncIterable
from typing import Any, cast
import azure.durable_functions as df
from agent_framework import AgentProtocol, AgentRunResponse, AgentRunResponseUpdate, get_logger
from agent_framework import AgentProtocol, AgentRunResponse, AgentRunResponseUpdate, Role, get_logger
from ._callbacks import AgentCallbackContext, AgentResponseCallbackProtocol
from ._models import AgentResponse, ChatRole, RunRequest
from ._models import AgentResponse, RunRequest
from ._state import AgentState
logger = get_logger("agent_framework.azurefunctions.entities")
@@ -81,33 +81,32 @@ class AgentEntity:
"""
# Convert string or dict to RunRequest
if isinstance(request, str):
run_request = RunRequest(message=request, role=ChatRole.USER)
run_request = RunRequest(message=request, role=Role.USER)
elif isinstance(request, dict):
run_request = RunRequest.from_dict(request)
else:
run_request = request
message = run_request.message
conversation_id = run_request.conversation_id
thread_id = run_request.thread_id
correlation_id = run_request.correlation_id
if not conversation_id:
raise ValueError("RunRequest must include a conversation_id")
if not thread_id:
raise ValueError("RunRequest must include a thread_id")
if not correlation_id:
raise ValueError("RunRequest must include a correlation_id")
role = run_request.role or ChatRole.USER
role = run_request.role or Role.USER
response_format = run_request.response_format
enable_tool_calls = run_request.enable_tool_calls
logger.debug(f"[AgentEntity.run_agent] Received message: {message}")
logger.debug(f"[AgentEntity.run_agent] Conversation ID: {conversation_id}")
logger.debug(f"[AgentEntity.run_agent] Thread ID: {thread_id}")
logger.debug(f"[AgentEntity.run_agent] Correlation ID: {correlation_id}")
logger.debug(f"[AgentEntity.run_agent] Role: {role.value if isinstance(role, ChatRole) else role}")
logger.debug(f"[AgentEntity.run_agent] Role: {role.value}")
logger.debug(f"[AgentEntity.run_agent] Enable tool calls: {enable_tool_calls}")
logger.debug(f"[AgentEntity.run_agent] Response format: {'provided' if response_format else 'none'}")
# Store message in history with role
role_str = role.value if isinstance(role, ChatRole) else role
self.state.add_user_message(message, role=role_str, correlation_id=correlation_id)
self.state.add_user_message(message, role=role, correlation_id=correlation_id)
logger.debug("[AgentEntity.run_agent] Executing agent...")
@@ -123,7 +122,7 @@ class AgentEntity:
agent_run_response: AgentRunResponse = await self._invoke_agent(
run_kwargs=run_kwargs,
correlation_id=correlation_id,
conversation_id=conversation_id,
thread_id=thread_id,
request_message=message,
)
@@ -160,7 +159,7 @@ class AgentEntity:
agent_response = AgentResponse(
response=response_text,
message=str(message),
conversation_id=str(conversation_id),
thread_id=str(thread_id),
status="success",
message_count=self.state.message_count,
structured_response=structured_response,
@@ -185,7 +184,7 @@ class AgentEntity:
error_response = AgentResponse(
response=f"Error: {exc!s}",
message=str(message),
conversation_id=str(conversation_id),
thread_id=str(thread_id),
status="error",
message_count=self.state.message_count,
error=str(exc),
@@ -197,7 +196,7 @@ class AgentEntity:
self,
run_kwargs: dict[str, Any],
correlation_id: str,
conversation_id: str,
thread_id: str,
request_message: str,
) -> AgentRunResponse:
"""Execute the agent, preferring streaming when available."""
@@ -205,7 +204,7 @@ class AgentEntity:
if self.callback is not None:
callback_context = self._build_callback_context(
correlation_id=correlation_id,
conversation_id=conversation_id,
thread_id=thread_id,
request_message=request_message,
)
@@ -319,7 +318,7 @@ class AgentEntity:
def _build_callback_context(
self,
correlation_id: str,
conversation_id: str,
thread_id: str,
request_message: str,
) -> AgentCallbackContext:
"""Create the callback context provided to consumers."""
@@ -327,7 +326,7 @@ class AgentEntity:
return AgentCallbackContext(
agent_name=agent_name,
correlation_id=correlation_id,
conversation_id=conversation_id,
thread_id=thread_id,
request_message=request_message,
)
@@ -375,11 +374,8 @@ def create_agent_entity(
if operation == "run_agent":
input_data: Any = context.get_input()
# Support both old format (message + conversation_id) and new format (RunRequest dict)
# This provides backward compatibility
request: str | dict[str, Any]
if isinstance(input_data, dict) and "message" in input_data:
# Input can be either old format or new RunRequest format
request = cast(dict[str, Any], input_data)
else:
# Fall back to treating input as message string
python/packages/azurefunctions/agent_framework_azurefunctions/_models.py
+52 -39
View File
@@ -5,21 +5,23 @@
This module defines the request and response models used by the framework.
"""
from __future__ import annotations
import inspect
import uuid
from collections.abc import MutableMapping
from dataclasses import dataclass
from enum import Enum
from importlib import import_module
from typing import TYPE_CHECKING, Any, cast
import azure.durable_functions as df
from agent_framework import AgentThread
from agent_framework import AgentThread, Role
if TYPE_CHECKING: # pragma: no cover - type checking imports only
from pydantic import BaseModel
_PydanticBaseModel: type["BaseModel"] | None
_PydanticBaseModel: type[BaseModel] | None
try:
from pydantic import BaseModel as _RuntimeBaseModel
except ImportError: # pragma: no cover - optional dependency
@@ -28,14 +30,6 @@ else:
_PydanticBaseModel = _RuntimeBaseModel
class ChatRole(str, Enum):
"""Chat message role enum."""
USER = "user"
SYSTEM = "system"
ASSISTANT = "assistant"
@dataclass
class AgentSessionId:
"""Represents an agent session ID, which is used to identify a long-running agent session.
@@ -63,7 +57,7 @@ class AgentSessionId:
return f"{AgentSessionId.ENTITY_NAME_PREFIX}{name}"
@staticmethod
def with_random_key(name: str) -> "AgentSessionId":
def with_random_key(name: str) -> AgentSessionId:
"""Creates a new AgentSessionId with the specified name and a randomly generated key.
Args:
@@ -83,7 +77,7 @@ class AgentSessionId:
return df.EntityId(self.to_entity_name(self.name), self.key)
@staticmethod
def from_entity_id(entity_id: df.EntityId) -> "AgentSessionId":
def from_entity_id(entity_id: df.EntityId) -> AgentSessionId:
"""Creates an AgentSessionId from a Durable Functions EntityId.
Args:
@@ -113,7 +107,7 @@ class AgentSessionId:
return f"AgentSessionId(name='{self.name}', key='{self.key}')"
@staticmethod
def parse(session_id_string: str) -> "AgentSessionId":
def parse(session_id_string: str) -> AgentSessionId:
"""Parses a string representation of an agent session ID.
Args:
@@ -172,7 +166,7 @@ class DurableAgentThread(AgentThread):
service_thread_id: str | None = None,
message_store: Any = None,
context_provider: Any = None,
) -> "DurableAgentThread":
) -> DurableAgentThread:
"""Creates a durable thread pre-associated with the supplied session ID."""
return cls(
session_id=session_id,
@@ -195,7 +189,7 @@ class DurableAgentThread(AgentThread):
*,
message_store: Any = None,
**kwargs: Any,
) -> "DurableAgentThread":
) -> DurableAgentThread:
"""Restores a durable thread, rehydrating the stored session identifier."""
state_payload = dict(serialized_thread_state)
session_id_value = state_payload.pop(cls._SERIALIZED_SESSION_ID_KEY, None)
@@ -217,7 +211,7 @@ class DurableAgentThread(AgentThread):
return thread
def _serialize_response_format(response_format: type["BaseModel"] | None) -> Any:
def _serialize_response_format(response_format: type[BaseModel] | None) -> Any:
"""Serialize response format for transport across durable function boundaries."""
if response_format is None:
return None
@@ -235,7 +229,7 @@ def _serialize_response_format(response_format: type["BaseModel"] | None) -> Any
}
def _deserialize_response_format(response_format: Any) -> type["BaseModel"] | None:
def _deserialize_response_format(response_format: Any) -> type[BaseModel] | None:
"""Deserialize response format back into actionable type if possible."""
if response_format is None:
return None
@@ -287,17 +281,45 @@ class RunRequest:
role: The role of the message sender (user, system, or assistant)
response_format: Optional Pydantic BaseModel type describing the structured response format
enable_tool_calls: Whether to enable tool calls for this request
conversation_id: Optional conversation/session ID for tracking
thread_id: Optional thread ID for tracking
correlation_id: Optional correlation ID for tracking the response to this specific request
"""
message: str
role: ChatRole = ChatRole.USER
response_format: type["BaseModel"] | None = None
role: Role = Role.USER
response_format: type[BaseModel] | None = None
enable_tool_calls: bool = True
conversation_id: str | None = None
thread_id: str | None = None
correlation_id: str | None = None
def __init__(
self,
message: str,
role: Role | str | None = Role.USER,
response_format: type[BaseModel] | None = None,
enable_tool_calls: bool = True,
thread_id: str | None = None,
correlation_id: str | None = None,
) -> None:
self.message = message
self.role = self.coerce_role(role)
self.response_format = response_format
self.enable_tool_calls = enable_tool_calls
self.thread_id = thread_id
self.correlation_id = correlation_id
@staticmethod
def coerce_role(value: Role | str | None) -> Role:
"""Normalize various role representations into a Role instance."""
if isinstance(value, Role):
return value
if isinstance(value, str):
normalized = value.strip()
if not normalized:
return Role.USER
return Role(value=normalized.lower())
return Role.USER
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for JSON serialization."""
result = {
@@ -307,30 +329,21 @@ class RunRequest:
}
if self.response_format:
result["response_format"] = _serialize_response_format(self.response_format)
if self.conversation_id:
result["conversation_id"] = self.conversation_id
if self.thread_id:
result["thread_id"] = self.thread_id
if self.correlation_id:
result["correlation_id"] = self.correlation_id
return result
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "RunRequest":
def from_dict(cls, data: dict[str, Any]) -> RunRequest:
"""Create RunRequest from dictionary."""
role_str = data.get("role")
if role_str:
try:
role = ChatRole(role_str.lower())
except ValueError:
role = ChatRole.USER # Default to USER if invalid
else:
role = ChatRole.USER
return cls(
message=data.get("message", ""),
role=role,
role=cls.coerce_role(data.get("role")),
response_format=_deserialize_response_format(data.get("response_format")),
enable_tool_calls=data.get("enable_tool_calls", True),
conversation_id=data.get("conversation_id"),
thread_id=data.get("thread_id"),
correlation_id=data.get("correlation_id"),
)
@@ -342,7 +355,7 @@ class AgentResponse:
Attributes:
response: The agent's text response (or None for structured responses)
message: The original message sent to the agent
conversation_id: The conversation/session ID
thread_id: The thread identifier
status: Status of the execution (success, error, etc.)
message_count: Number of messages in the conversation
error: Error message if status is error
@@ -352,7 +365,7 @@ class AgentResponse:
response: str | None
message: str
conversation_id: str | None
thread_id: str | None
status: str
message_count: int = 0
error: str | None = None
@@ -363,7 +376,7 @@ class AgentResponse:
"""Convert to dictionary for JSON serialization."""
result = {
"message": self.message,
"conversation_id": self.conversation_id,
"thread_id": self.thread_id,
"status": self.status,
"message_count": self.message_count,
}
python/packages/azurefunctions/agent_framework_azurefunctions/_orchestration.py
+1 -1
View File
@@ -136,7 +136,7 @@ class DurableAIAgent(AgentProtocol):
message=message_str,
enable_tool_calls=enable_tool_calls,
correlation_id=correlation_id,
conversation_id=session_id.key,
thread_id=session_id.key,
response_format=response_format,
)
python/packages/azurefunctions/agent_framework_azurefunctions/_state.py
+3 -3
View File
@@ -8,9 +8,9 @@ serializing agent framework responses.
from collections.abc import MutableMapping
from datetime import datetime, timezone
from typing import Any, Literal, cast
from typing import Any, cast
from agent_framework import AgentRunResponse, ChatMessage, get_logger
from agent_framework import AgentRunResponse, ChatMessage, Role, get_logger
logger = get_logger("agent_framework.azurefunctions.state")
@@ -38,7 +38,7 @@ class AgentState:
def add_user_message(
self,
content: str,
role: Literal["user", "system", "assistant", "tool"] = "user",
role: Role = Role.USER,
correlation_id: str | None = None,
) -> None:
"""Add a user message to the conversation history as a ChatMessage object.
python/packages/azurefunctions/tests/integration_tests/test_01_single_agent.py
+15 -17
View File
@@ -45,9 +45,9 @@ class TestSampleSingleAgent:
"""Test sending a simple message with JSON payload."""
response = SampleTestHelper.post_json(
f"{self.base_url}/run",
{"message": "Tell me a short joke about cloud computing.", "sessionId": "test-simple-json"},
{"message": "Tell me a short joke about cloud computing.", "thread_id": "test-simple-json"},
)
# Agent can return 200 (immediate) or 202 (async with wait_for_completion=false)
# Agent can return 200 (immediate) or 202 (async with wait_for_response=false)
assert response.status_code in [200, 202]
data = response.json()
@@ -58,37 +58,35 @@ class TestSampleSingleAgent:
assert data["message_count"] >= 1
else:
# Async response - check we got correlation info
assert "correlationId" in data or "sessionId" in data
assert "correlation_id" in data or "thread_id" in data
def test_simple_message_plain_text(self) -> None:
"""Test sending a message with plain text payload."""
response = SampleTestHelper.post_text(f"{self.base_url}/run", "Tell me a short joke about networking.")
assert response.status_code in [200, 202]
data = response.json()
if response.status_code == 200:
assert data["status"] == "success"
assert "response" in data
# Agent responded with plain text when the request body was text/plain.
assert response.text.strip()
assert response.headers.get("x-ms-thread-id") is not None
def test_session_key_in_query(self) -> None:
"""Test using sessionKey in query parameter."""
def test_thread_id_in_query(self) -> None:
"""Test using thread_id in query parameter."""
response = SampleTestHelper.post_text(
f"{self.base_url}/run?sessionKey=test-query-session", "Tell me a short joke about weather in Texas."
f"{self.base_url}/run?thread_id=test-query-thread", "Tell me a short joke about weather in Texas."
)
assert response.status_code in [200, 202]
data = response.json()
if response.status_code == 200:
assert data["status"] == "success"
assert response.text.strip()
assert response.headers.get("x-ms-thread-id") == "test-query-thread"
def test_conversation_continuity(self) -> None:
"""Test conversation context is maintained across requests."""
session_id = "test-continuity"
thread_id = "test-continuity"
# First message
response1 = SampleTestHelper.post_json(
f"{self.base_url}/run",
{"message": "Tell me a short joke about weather in Seattle.", "sessionId": session_id},
{"message": "Tell me a short joke about weather in Seattle.", "thread_id": thread_id},
)
assert response1.status_code in [200, 202]
@@ -98,7 +96,7 @@ class TestSampleSingleAgent:
# Second message in same session
response2 = SampleTestHelper.post_json(
f"{self.base_url}/run", {"message": "What about San Francisco?", "sessionId": session_id}
f"{self.base_url}/run", {"message": "What about San Francisco?", "thread_id": thread_id}
)
assert response2.status_code == 200
data2 = response2.json()
@@ -107,7 +105,7 @@ class TestSampleSingleAgent:
# In async mode, we can't easily test message count
# Just verify we can make multiple calls
response2 = SampleTestHelper.post_json(
f"{self.base_url}/run", {"message": "What about Texas?", "sessionId": session_id}
f"{self.base_url}/run", {"message": "What about Texas?", "thread_id": thread_id}
)
assert response2.status_code == 202
python/packages/azurefunctions/tests/integration_tests/test_02_multi_agent.py
+10 -5
View File
@@ -38,21 +38,26 @@ class TestSampleMultiAgent:
def test_weather_agent(self) -> None:
"""Test WeatherAgent endpoint."""
response = SampleTestHelper.post_json(
f"{self.weather_base_url}/run", {"message": "What is the weather in Seattle?"}
f"{self.weather_base_url}/run",
{"message": "What is the weather in Seattle?"},
)
assert response.status_code == 202
assert response.status_code == 200
data = response.json()
assert data["status"] == "accepted"
assert data["status"] == "success"
assert "response" in data
def test_math_agent(self) -> None:
"""Test MathAgent endpoint."""
response = SampleTestHelper.post_json(
f"{self.math_base_url}/run", {"message": "Calculate a 20% tip on a $50 bill"}
f"{self.math_base_url}/run",
{"message": "Calculate a 20% tip on a $50 bill", "wait_for_response": False},
)
assert response.status_code == 202
data = response.json()
assert data["status"] == "accepted"
assert "response" in data
assert "correlation_id" in data
assert "thread_id" in data
if __name__ == "__main__":
python/packages/azurefunctions/tests/integration_tests/test_03_callbacks.py
+33 -10
View File
@@ -14,6 +14,8 @@ Usage:
uv run pytest packages/azurefunctions/tests/integration_tests/test_03_callbacks.py -v
"""
from typing import Any
import pytest
import requests
@@ -39,39 +41,60 @@ class TestSampleCallbacks:
"""Provide the callback agent base URL for each test."""
self.base_url = f"{base_url}/api/agents/CallbackAgent"
@staticmethod
def _wait_for_callback_events(base_url: str, thread_id: str) -> list[dict[str, Any]]:
events: list[dict[str, Any]] = []
response = SampleTestHelper.get(f"{base_url}/callbacks/{thread_id}")
if response.status_code == 200:
events = response.json()
return events
def test_agent_with_callbacks(self) -> None:
"""Test agent execution with callback tracking."""
conversation_id = "test-callback"
thread_id = "test-callback"
response = SampleTestHelper.post_json(
f"{self.base_url}/run", {"message": "Tell me about Python", "conversationId": conversation_id}
f"{self.base_url}/run",
{"message": "Tell me about Python", "thread_id": thread_id},
)
assert response.status_code == 202
assert response.status_code == 200
data = response.json()
assert data["status"] == "accepted"
assert data["status"] == "success"
events = self._wait_for_callback_events(self.base_url, thread_id)
assert events
assert any(event.get("event_type") == "final" for event in events)
def test_get_callbacks(self) -> None:
"""Test retrieving callback events."""
conversation_id = "test-callback-retrieve"
thread_id = "test-callback-retrieve"
# Send a message first
SampleTestHelper.post_json(f"{self.base_url}/run", {"message": "Hello", "conversationId": conversation_id})
SampleTestHelper.post_json(
f"{self.base_url}/run",
{"message": "Hello", "thread_id": thread_id, "wait_for_response": False},
)
# Get callbacks
response = SampleTestHelper.get(f"{self.base_url}/callbacks/{conversation_id}")
response = SampleTestHelper.get(f"{self.base_url}/callbacks/{thread_id}")
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
def test_delete_callbacks(self) -> None:
"""Test clearing callback events."""
conversation_id = "test-callback-delete"
thread_id = "test-callback-delete"
# Send a message first
SampleTestHelper.post_json(f"{self.base_url}/run", {"message": "Test", "conversationId": conversation_id})
SampleTestHelper.post_json(
f"{self.base_url}/run",
{"message": "Test", "thread_id": thread_id, "wait_for_response": False},
)
# Delete callbacks
response = requests.delete(f"{self.base_url}/callbacks/{conversation_id}", timeout=TIMEOUT)
response = requests.delete(f"{self.base_url}/callbacks/{thread_id}", timeout=TIMEOUT)
assert response.status_code == 204
python/packages/azurefunctions/tests/test_app.py
+163 -39
View File
@@ -12,8 +12,8 @@ import pytest
from agent_framework import AgentRunResponse, ChatMessage
from agent_framework_azurefunctions import AgentFunctionApp
from agent_framework_azurefunctions._app import WAIT_FOR_RESPONSE_FIELD, WAIT_FOR_RESPONSE_HEADER
from agent_framework_azurefunctions._entities import AgentEntity, AgentState, create_agent_entity
from agent_framework_azurefunctions._errors import IncomingRequestError
TFunc = TypeVar("TFunc", bound=Callable[..., Any])
@@ -150,6 +150,38 @@ class TestAgentFunctionAppSetup:
# Verify agent is registered
assert "TestAgent" in app.agents
def test_http_function_name_uses_prefix_format(self) -> None:
"""Ensure function names follow the prefix-agent naming convention."""
mock_agent = Mock()
mock_agent.name = "Agent 42"
captured_names: list[str] = []
def capture_function_name(
self: AgentFunctionApp, name: str, *args: Any, **kwargs: Any
) -> Callable[[TFunc], TFunc]:
def decorator(func: TFunc) -> TFunc:
captured_names.append(name)
return func
return decorator
def passthrough_decorator(*args: Any, **kwargs: Any) -> Callable[[TFunc], TFunc]:
def decorator(func: TFunc) -> TFunc:
return func
return decorator
with (
patch.object(AgentFunctionApp, "function_name", new=capture_function_name),
patch.object(AgentFunctionApp, "route", new=passthrough_decorator),
patch.object(AgentFunctionApp, "durable_client_input", new=passthrough_decorator),
patch.object(AgentFunctionApp, "entity_trigger", new=passthrough_decorator),
):
AgentFunctionApp(agents=[mock_agent])
assert captured_names == ["http-Agent_42"]
def test_setup_skips_http_trigger_when_disabled(self) -> None:
"""Test that HTTP trigger is not created when disabled."""
mock_agent = Mock()
@@ -236,8 +268,8 @@ class TestAgentFunctionAppSetup:
assert "Agent2" in app2.agents
class TestWaitForCompletionAndCorrelationId:
"""Tests for wait_for_completion flag and correlation ID handling."""
class TestWaitForResponseAndCorrelationId:
"""Tests for wait_for_response flag and correlation ID handling."""
def _create_app(self) -> AgentFunctionApp:
mock_agent = Mock()
@@ -255,21 +287,35 @@ class TestWaitForCompletionAndCorrelationId:
request.params = params or {}
return request
def test_wait_for_completion_header_true(self) -> None:
"""Test that the wait-for-completion header is honored."""
def test_wait_for_response_header_true(self) -> None:
"""Test that the wait-for-response header is honored."""
app = self._create_app()
request = self._make_request(headers={"X-Wait-For-Completion": "true"})
request = self._make_request(headers={WAIT_FOR_RESPONSE_HEADER: "true"})
assert app._should_wait_for_completion(request, {}) is True
assert app._should_wait_for_response(request, {}) is True
def test_wait_for_completion_body_variants(self) -> None:
"""Test that multiple payload spellings are accepted."""
def test_wait_for_response_body_snake_case(self) -> None:
"""Test that payload controls wait_for_response."""
app = self._create_app()
request = self._make_request()
assert app._should_wait_for_completion(request, {"wait_for_completion": "true"}) is True
assert app._should_wait_for_completion(request, {"waitForCompletion": "1"}) is True
assert app._should_wait_for_completion(request, {"WaitForCompletion": "no"}) is False
assert app._should_wait_for_response(request, {WAIT_FOR_RESPONSE_FIELD: "true"}) is True
assert app._should_wait_for_response(request, {WAIT_FOR_RESPONSE_FIELD: "false"}) is False
assert app._should_wait_for_response(request, {WAIT_FOR_RESPONSE_FIELD: "0"}) is False
def test_wait_for_response_query_parameter(self) -> None:
"""Test that query parameter controls wait_for_response."""
app = self._create_app()
request = self._make_request(params={WAIT_FOR_RESPONSE_FIELD: "true"})
assert app._should_wait_for_response(request, {}) is True
def test_wait_for_response_query_precedence(self) -> None:
"""Test that query parameter overrides body value."""
app = self._create_app()
request = self._make_request(params={WAIT_FOR_RESPONSE_FIELD: "false"})
assert app._should_wait_for_response(request, {WAIT_FOR_RESPONSE_FIELD: "true"}) is False
class TestAgentEntityOperations:
@@ -287,13 +333,13 @@ class TestAgentEntityOperations:
result = await entity.run_agent(
mock_context,
{"message": "Test message", "conversation_id": "test-conv-123", "correlation_id": "corr-app-entity-1"},
{"message": "Test message", "thread_id": "test-conv-123", "correlation_id": "corr-app-entity-1"},
)
assert result["status"] == "success"
assert result["response"] == "Test response"
assert result["message"] == "Test message"
assert result["conversation_id"] == "test-conv-123"
assert result["thread_id"] == "test-conv-123"
assert entity.state.message_count == 1
async def test_entity_stores_conversation_history(self) -> None:
@@ -308,7 +354,7 @@ class TestAgentEntityOperations:
# Send first message
await entity.run_agent(
mock_context, {"message": "Message 1", "conversation_id": "conv-1", "correlation_id": "corr-app-entity-2"}
mock_context, {"message": "Message 1", "thread_id": "conv-1", "correlation_id": "corr-app-entity-2"}
)
history = entity.state.conversation_history
@@ -337,12 +383,12 @@ class TestAgentEntityOperations:
assert entity.state.message_count == 0
await entity.run_agent(
mock_context, {"message": "Message 1", "conversation_id": "conv-1", "correlation_id": "corr-app-entity-3a"}
mock_context, {"message": "Message 1", "thread_id": "conv-1", "correlation_id": "corr-app-entity-3a"}
)
assert entity.state.message_count == 1
await entity.run_agent(
mock_context, {"message": "Message 2", "conversation_id": "conv-1", "correlation_id": "corr-app-entity-3b"}
mock_context, {"message": "Message 2", "thread_id": "conv-1", "correlation_id": "corr-app-entity-3b"}
)
assert entity.state.message_count == 2
@@ -391,7 +437,7 @@ class TestAgentEntityFactory:
mock_context.operation_name = "run_agent"
mock_context.get_input.return_value = {
"message": "Test message",
"conversation_id": "conv-123",
"thread_id": "conv-123",
"correlation_id": "corr-app-factory-1",
}
mock_context.get_state.return_value = None
@@ -478,7 +524,7 @@ class TestErrorHandling:
mock_context = Mock()
result = await entity.run_agent(
mock_context, {"message": "Test message", "conversation_id": "conv-1", "correlation_id": "corr-app-error-1"}
mock_context, {"message": "Test message", "thread_id": "conv-1", "correlation_id": "corr-app-error-1"}
)
assert result["status"] == "error"
@@ -521,48 +567,67 @@ class TestIncomingRequestParsing:
app = self._create_app()
request = Mock()
request.headers = {}
request.params = {}
request.get_json.side_effect = ValueError("Invalid JSON")
request.get_body.return_value = b"Plain text message"
req_body, message = app._parse_incoming_request(request)
req_body, message, response_format = app._parse_incoming_request(request)
assert req_body == {}
assert message == "Plain text message"
def test_parse_plain_text_requires_content(self) -> None:
"""Test that plain-text requests require message content."""
assert response_format == "text"
def test_parse_plain_text_trims_whitespace(self) -> None:
"""Plain-text parser returns an empty string when the body contains only whitespace."""
app = self._create_app()
request = Mock()
request.headers = {}
request.params = {}
request.get_json.side_effect = ValueError("Invalid JSON")
request.get_body.return_value = b" "
with pytest.raises(IncomingRequestError) as exc_info:
app._parse_incoming_request(request)
req_body, message, response_format = app._parse_incoming_request(request)
assert "Message is required" in str(exc_info.value)
assert req_body == {}
assert message == ""
assert response_format == "text"
def test_extract_session_key_from_query_params(self) -> None:
"""Test session key extraction from query parameters."""
def test_accept_header_prefers_json(self) -> None:
"""Test that the Accept header can force JSON responses for plain-text bodies."""
app = self._create_app()
request = Mock()
request.params = {"sessionId": "query-session"}
request.headers = {"accept": "application/json"}
request.params = {}
request.get_json.side_effect = ValueError("Invalid JSON")
request.get_body.return_value = b"Plain text message"
_, message, response_format = app._parse_incoming_request(request)
assert message == "Plain text message"
assert response_format == "json"
def test_extract_thread_id_from_query_params(self) -> None:
"""Test thread identifier extraction from query parameters."""
app = self._create_app()
request = Mock()
request.params = {"thread_id": "query-thread"}
req_body = {}
session_key = app._resolve_session_key(request, req_body)
thread_id = app._resolve_thread_id(request, req_body)
assert session_key == "query-session"
assert thread_id == "query-thread"
class TestHttpRunRoute:
"""Tests for the HTTP run route behavior."""
async def test_http_run_accepts_plain_text(self) -> None:
"""Test that the HTTP handler accepts plain-text requests."""
mock_agent = Mock()
mock_agent.name = "HttpAgent"
@staticmethod
def _get_run_handler(agent: Mock) -> Callable[[func.HttpRequest, Any], Awaitable[func.HttpResponse]]:
captured_handlers: dict[str | None, Callable[..., Awaitable[func.HttpResponse]]] = {}
def capture_decorator(*args: Any, **kwargs: Any) -> Callable[[TFunc], TFunc]:
@@ -585,13 +650,20 @@ class TestHttpRunRoute:
patch.object(AgentFunctionApp, "durable_client_input", new=capture_decorator),
patch.object(AgentFunctionApp, "entity_trigger", new=capture_decorator),
):
AgentFunctionApp(agents=[mock_agent], enable_health_check=False)
AgentFunctionApp(agents=[agent], enable_health_check=False)
run_route = f"agents/{mock_agent.name}/run"
handler = captured_handlers[run_route]
run_route = f"agents/{agent.name}/run"
return captured_handlers[run_route]
async def test_http_run_accepts_plain_text(self) -> None:
"""Test that the HTTP handler accepts plain-text requests."""
mock_agent = Mock()
mock_agent.name = "HttpAgent"
handler = self._get_run_handler(mock_agent)
request = Mock()
request.headers = {}
request.headers = {WAIT_FOR_RESPONSE_HEADER: "false"}
request.params = {}
request.route_params = {}
request.get_json.side_effect = ValueError("Invalid JSON")
@@ -602,12 +674,64 @@ class TestHttpRunRoute:
response = await handler(request, client)
assert response.status_code == 202
assert response.mimetype == "text/plain"
assert response.headers.get("x-ms-thread-id") is not None
assert response.get_body().decode("utf-8") == "Agent request accepted"
signal_args = client.signal_entity.call_args[0]
run_request = signal_args[2]
assert run_request["message"] == "Plain text via HTTP"
assert run_request["role"] == "user"
assert "thread_id" in run_request
async def test_http_run_accept_header_returns_json(self) -> None:
"""Test that Accept header requesting JSON results in JSON response."""
mock_agent = Mock()
mock_agent.name = "HttpAgentJson"
handler = self._get_run_handler(mock_agent)
request = Mock()
request.headers = {WAIT_FOR_RESPONSE_HEADER: "false", "Accept": "application/json"}
request.params = {}
request.route_params = {}
request.get_json.side_effect = ValueError("Invalid JSON")
request.get_body.return_value = b"Plain text via HTTP"
client = AsyncMock()
response = await handler(request, client)
assert response.status_code == 202
assert response.mimetype == "application/json"
assert response.headers.get("x-ms-thread-id") is None
body = response.get_body().decode("utf-8")
assert '"status": "accepted"' in body
async def test_http_run_rejects_empty_message(self) -> None:
"""Test that the HTTP handler rejects empty messages with a 400 response."""
mock_agent = Mock()
mock_agent.name = "HttpAgentEmpty"
handler = self._get_run_handler(mock_agent)
request = Mock()
request.headers = {WAIT_FOR_RESPONSE_HEADER: "false"}
request.params = {}
request.route_params = {}
request.get_json.side_effect = ValueError("Invalid JSON")
request.get_body.return_value = b" "
client = AsyncMock()
response = await handler(request, client)
assert response.status_code == 400
assert response.mimetype == "text/plain"
assert response.headers.get("x-ms-thread-id") is not None
assert response.get_body().decode("utf-8") == "Message is required"
client.signal_entity.assert_not_called()
if __name__ == "__main__":
python/packages/azurefunctions/tests/test_entities.py
+48 -48
View File
@@ -12,11 +12,11 @@ from typing import Any, TypeVar
from unittest.mock import AsyncMock, Mock, patch
import pytest
from agent_framework import AgentRunResponse, AgentRunResponseUpdate, ChatMessage
from agent_framework import AgentRunResponse, AgentRunResponseUpdate, ChatMessage, Role
from pydantic import BaseModel
from agent_framework_azurefunctions._entities import AgentEntity, create_agent_entity
from agent_framework_azurefunctions._models import ChatRole, RunRequest
from agent_framework_azurefunctions._models import RunRequest
from agent_framework_azurefunctions._state import AgentState
TFunc = TypeVar("TFunc", bound=Callable[..., Any])
@@ -112,7 +112,7 @@ class TestAgentEntityRunAgent:
mock_context = Mock()
result = await entity.run_agent(
mock_context, {"message": "Test message", "conversation_id": "conv-123", "correlation_id": "corr-entity-1"}
mock_context, {"message": "Test message", "thread_id": "conv-123", "correlation_id": "corr-entity-1"}
)
# Verify agent.run was called
@@ -130,7 +130,7 @@ class TestAgentEntityRunAgent:
assert result["status"] == "success"
assert result["response"] == "Test response"
assert result["message"] == "Test message"
assert result["conversation_id"] == "conv-123"
assert result["thread_id"] == "conv-123"
async def test_run_agent_streaming_callbacks_invoked(self) -> None:
"""Ensure streaming updates trigger callbacks and run() is not used."""
@@ -157,7 +157,7 @@ class TestAgentEntityRunAgent:
mock_context,
{
"message": "Tell me something",
"conversation_id": "session-1",
"thread_id": "session-1",
"correlation_id": "corr-stream-1",
},
)
@@ -175,7 +175,7 @@ class TestAgentEntityRunAgent:
context = recorded_call.args[1]
assert context.agent_name == "StreamingAgent"
assert context.correlation_id == "corr-stream-1"
assert context.conversation_id == "session-1"
assert context.thread_id == "session-1"
assert context.request_message == "Tell me something"
final_call = callback.response_mock.await_args
@@ -183,7 +183,7 @@ class TestAgentEntityRunAgent:
final_response, final_context = final_call.args
assert final_context.agent_name == "StreamingAgent"
assert final_context.correlation_id == "corr-stream-1"
assert final_context.conversation_id == "session-1"
assert final_context.thread_id == "session-1"
assert final_context.request_message == "Tell me something"
assert getattr(final_response, "text", "").strip()
@@ -204,7 +204,7 @@ class TestAgentEntityRunAgent:
mock_context,
{
"message": "Hi",
"conversation_id": "session-2",
"thread_id": "session-2",
"correlation_id": "corr-final-1",
},
)
@@ -220,7 +220,7 @@ class TestAgentEntityRunAgent:
final_context = final_call.args[1]
assert final_context.agent_name == "NonStreamingAgent"
assert final_context.correlation_id == "corr-final-1"
assert final_context.conversation_id == "session-2"
assert final_context.thread_id == "session-2"
assert final_context.request_message == "Hi"
async def test_run_agent_updates_conversation_history(self) -> None:
@@ -233,7 +233,7 @@ class TestAgentEntityRunAgent:
mock_context = Mock()
await entity.run_agent(
mock_context, {"message": "User message", "conversation_id": "conv-1", "correlation_id": "corr-entity-2"}
mock_context, {"message": "User message", "thread_id": "conv-1", "correlation_id": "corr-entity-2"}
)
# Should have 2 entries: user message + assistant response
@@ -260,17 +260,17 @@ class TestAgentEntityRunAgent:
assert entity.state.message_count == 0
await entity.run_agent(
mock_context, {"message": "Message 1", "conversation_id": "conv-1", "correlation_id": "corr-entity-3a"}
mock_context, {"message": "Message 1", "thread_id": "conv-1", "correlation_id": "corr-entity-3a"}
)
assert entity.state.message_count == 1
await entity.run_agent(
mock_context, {"message": "Message 2", "conversation_id": "conv-1", "correlation_id": "corr-entity-3b"}
mock_context, {"message": "Message 2", "thread_id": "conv-1", "correlation_id": "corr-entity-3b"}
)
assert entity.state.message_count == 2
await entity.run_agent(
mock_context, {"message": "Message 3", "conversation_id": "conv-1", "correlation_id": "corr-entity-3c"}
mock_context, {"message": "Message 3", "thread_id": "conv-1", "correlation_id": "corr-entity-3c"}
)
assert entity.state.message_count == 3
@@ -283,27 +283,27 @@ class TestAgentEntityRunAgent:
mock_context = Mock()
await entity.run_agent(
mock_context, {"message": "Message 1", "conversation_id": "conv-1", "correlation_id": "corr-entity-4a"}
mock_context, {"message": "Message 1", "thread_id": "conv-1", "correlation_id": "corr-entity-4a"}
)
assert entity.state.last_response == "Response 1"
mock_agent.run = AsyncMock(return_value=_agent_response("Response 2"))
await entity.run_agent(
mock_context, {"message": "Message 2", "conversation_id": "conv-1", "correlation_id": "corr-entity-4b"}
mock_context, {"message": "Message 2", "thread_id": "conv-1", "correlation_id": "corr-entity-4b"}
)
assert entity.state.last_response == "Response 2"
async def test_run_agent_with_none_conversation_id(self) -> None:
"""Test run_agent with a None conversation identifier."""
async def test_run_agent_with_none_thread_id(self) -> None:
"""Test run_agent with a None thread identifier."""
mock_agent = Mock()
mock_agent.run = AsyncMock(return_value=_agent_response("Response"))
entity = AgentEntity(mock_agent)
mock_context = Mock()
with pytest.raises(ValueError, match="conversation_id"):
with pytest.raises(ValueError, match="thread_id"):
await entity.run_agent(
mock_context, {"message": "Message", "conversation_id": None, "correlation_id": "corr-entity-5"}
mock_context, {"message": "Message", "thread_id": None, "correlation_id": "corr-entity-5"}
)
async def test_run_agent_handles_response_without_text_attribute(self) -> None:
@@ -322,7 +322,7 @@ class TestAgentEntityRunAgent:
mock_context = Mock()
result = await entity.run_agent(
mock_context, {"message": "Message", "conversation_id": "conv-1", "correlation_id": "corr-entity-6"}
mock_context, {"message": "Message", "thread_id": "conv-1", "correlation_id": "corr-entity-6"}
)
# Should handle gracefully
@@ -338,7 +338,7 @@ class TestAgentEntityRunAgent:
mock_context = Mock()
result = await entity.run_agent(
mock_context, {"message": "Message", "conversation_id": "conv-1", "correlation_id": "corr-entity-7"}
mock_context, {"message": "Message", "thread_id": "conv-1", "correlation_id": "corr-entity-7"}
)
assert result["status"] == "success"
@@ -354,13 +354,13 @@ class TestAgentEntityRunAgent:
# Send multiple messages
await entity.run_agent(
mock_context, {"message": "Message 1", "conversation_id": "conv-1", "correlation_id": "corr-entity-8a"}
mock_context, {"message": "Message 1", "thread_id": "conv-1", "correlation_id": "corr-entity-8a"}
)
await entity.run_agent(
mock_context, {"message": "Message 2", "conversation_id": "conv-1", "correlation_id": "corr-entity-8b"}
mock_context, {"message": "Message 2", "thread_id": "conv-1", "correlation_id": "corr-entity-8b"}
)
await entity.run_agent(
mock_context, {"message": "Message 3", "conversation_id": "conv-1", "correlation_id": "corr-entity-8c"}
mock_context, {"message": "Message 3", "thread_id": "conv-1", "correlation_id": "corr-entity-8c"}
)
history = entity.state.conversation_history
@@ -421,10 +421,10 @@ class TestAgentEntityReset:
# Have a conversation
await entity.run_agent(
mock_context, {"message": "Message 1", "conversation_id": "conv-1", "correlation_id": "corr-entity-10a"}
mock_context, {"message": "Message 1", "thread_id": "conv-1", "correlation_id": "corr-entity-10a"}
)
await entity.run_agent(
mock_context, {"message": "Message 2", "conversation_id": "conv-1", "correlation_id": "corr-entity-10b"}
mock_context, {"message": "Message 2", "thread_id": "conv-1", "correlation_id": "corr-entity-10b"}
)
# Verify state before reset
@@ -463,7 +463,7 @@ class TestCreateAgentEntity:
mock_context.operation_name = "run_agent"
mock_context.get_input.return_value = {
"message": "Test message",
"conversation_id": "conv-123",
"thread_id": "conv-123",
"correlation_id": "corr-entity-factory",
}
mock_context.get_state.return_value = None
@@ -591,7 +591,7 @@ class TestErrorHandling:
mock_context = Mock()
result = await entity.run_agent(
mock_context, {"message": "Message", "conversation_id": "conv-1", "correlation_id": "corr-entity-error-1"}
mock_context, {"message": "Message", "thread_id": "conv-1", "correlation_id": "corr-entity-error-1"}
)
assert result["status"] == "error"
@@ -608,7 +608,7 @@ class TestErrorHandling:
mock_context = Mock()
result = await entity.run_agent(
mock_context, {"message": "Message", "conversation_id": "conv-1", "correlation_id": "corr-entity-error-2"}
mock_context, {"message": "Message", "thread_id": "conv-1", "correlation_id": "corr-entity-error-2"}
)
assert result["status"] == "error"
@@ -624,7 +624,7 @@ class TestErrorHandling:
mock_context = Mock()
result = await entity.run_agent(
mock_context, {"message": "Message", "conversation_id": "conv-1", "correlation_id": "corr-entity-error-3"}
mock_context, {"message": "Message", "thread_id": "conv-1", "correlation_id": "corr-entity-error-3"}
)
assert result["status"] == "error"
@@ -659,12 +659,12 @@ class TestErrorHandling:
result = await entity.run_agent(
mock_context,
{"message": "Test message", "conversation_id": "conv-123", "correlation_id": "corr-entity-error-4"},
{"message": "Test message", "thread_id": "conv-123", "correlation_id": "corr-entity-error-4"},
)
# Even on error, message info should be preserved
assert result["message"] == "Test message"
assert result["conversation_id"] == "conv-123"
assert result["thread_id"] == "conv-123"
assert result["status"] == "error"
@@ -680,7 +680,7 @@ class TestConversationHistory:
mock_context = Mock()
await entity.run_agent(
mock_context, {"message": "Message", "conversation_id": "conv-1", "correlation_id": "corr-entity-history-1"}
mock_context, {"message": "Message", "thread_id": "conv-1", "correlation_id": "corr-entity-history-1"}
)
# Check both user and assistant messages have timestamps
@@ -701,19 +701,19 @@ class TestConversationHistory:
mock_agent.run = AsyncMock(return_value=_agent_response("Response 1"))
await entity.run_agent(
mock_context,
{"message": "Message 1", "conversation_id": "conv-1", "correlation_id": "corr-entity-history-2a"},
{"message": "Message 1", "thread_id": "conv-1", "correlation_id": "corr-entity-history-2a"},
)
mock_agent.run = AsyncMock(return_value=_agent_response("Response 2"))
await entity.run_agent(
mock_context,
{"message": "Message 2", "conversation_id": "conv-1", "correlation_id": "corr-entity-history-2b"},
{"message": "Message 2", "thread_id": "conv-1", "correlation_id": "corr-entity-history-2b"},
)
mock_agent.run = AsyncMock(return_value=_agent_response("Response 3"))
await entity.run_agent(
mock_context,
{"message": "Message 3", "conversation_id": "conv-1", "correlation_id": "corr-entity-history-2c"},
{"message": "Message 3", "thread_id": "conv-1", "correlation_id": "corr-entity-history-2c"},
)
# Verify order
@@ -735,11 +735,11 @@ class TestConversationHistory:
await entity.run_agent(
mock_context,
{"message": "Message 1", "conversation_id": "conv-1", "correlation_id": "corr-entity-history-3a"},
{"message": "Message 1", "thread_id": "conv-1", "correlation_id": "corr-entity-history-3a"},
)
await entity.run_agent(
mock_context,
{"message": "Message 2", "conversation_id": "conv-1", "correlation_id": "corr-entity-history-3b"},
{"message": "Message 2", "thread_id": "conv-1", "correlation_id": "corr-entity-history-3b"},
)
# Check role alternation
@@ -763,8 +763,8 @@ class TestRunRequestSupport:
request = RunRequest(
message="Test message",
conversation_id="conv-123",
role=ChatRole.USER,
thread_id="conv-123",
role=Role.USER,
enable_tool_calls=True,
correlation_id="corr-runreq-1",
)
@@ -774,7 +774,7 @@ class TestRunRequestSupport:
assert result["status"] == "success"
assert result["response"] == "Response"
assert result["message"] == "Test message"
assert result["conversation_id"] == "conv-123"
assert result["thread_id"] == "conv-123"
async def test_run_agent_with_dict_request(self) -> None:
"""Test run_agent with a dictionary request."""
@@ -786,7 +786,7 @@ class TestRunRequestSupport:
request_dict = {
"message": "Test message",
"conversation_id": "conv-456",
"thread_id": "conv-456",
"role": "system",
"enable_tool_calls": False,
"correlation_id": "corr-runreq-2",
@@ -796,7 +796,7 @@ class TestRunRequestSupport:
assert result["status"] == "success"
assert result["message"] == "Test message"
assert result["conversation_id"] == "conv-456"
assert result["thread_id"] == "conv-456"
async def test_run_agent_with_string_raises_without_correlation(self) -> None:
"""Test that run_agent rejects legacy string input without correlation ID."""
@@ -820,8 +820,8 @@ class TestRunRequestSupport:
# Send as system role
request = RunRequest(
message="System message",
conversation_id="conv-runreq-3",
role=ChatRole.SYSTEM,
thread_id="conv-runreq-3",
role=Role.SYSTEM,
correlation_id="corr-runreq-3",
)
@@ -843,7 +843,7 @@ class TestRunRequestSupport:
request = RunRequest(
message="What is the answer?",
conversation_id="conv-runreq-4",
thread_id="conv-runreq-4",
response_format=EntityStructuredResponse,
correlation_id="corr-runreq-4",
)
@@ -864,7 +864,7 @@ class TestRunRequestSupport:
mock_context = Mock()
request = RunRequest(
message="Test", conversation_id="conv-runreq-5", enable_tool_calls=False, correlation_id="corr-runreq-5"
message="Test", thread_id="conv-runreq-5", enable_tool_calls=False, correlation_id="corr-runreq-5"
)
result = await entity.run_agent(mock_context, request)
@@ -884,7 +884,7 @@ class TestRunRequestSupport:
mock_context.operation_name = "run_agent"
mock_context.get_input.return_value = {
"message": "Test message",
"conversation_id": "conv-789",
"thread_id": "conv-789",
"role": "user",
"enable_tool_calls": True,
"correlation_id": "corr-runreq-6",
python/packages/azurefunctions/tests/test_models.py
+61 -69
View File
@@ -1,34 +1,19 @@
# Copyright (c) Microsoft. All rights reserved.
"""Unit tests for data models (AgentSessionId, RunRequest, AgentResponse, ChatRole)."""
"""Unit tests for data models (AgentSessionId, RunRequest, AgentResponse)."""
import azure.durable_functions as df
import pytest
from agent_framework import Role
from pydantic import BaseModel
from agent_framework_azurefunctions._models import AgentResponse, AgentSessionId, ChatRole, RunRequest
from agent_framework_azurefunctions._models import AgentResponse, AgentSessionId, RunRequest
class ModuleStructuredResponse(BaseModel):
value: int
class TestChatRole:
"""Test suite for ChatRole enum."""
def test_chat_role_values(self) -> None:
"""Test that ChatRole has correct values."""
assert ChatRole.USER == "user"
assert ChatRole.SYSTEM == "system"
assert ChatRole.ASSISTANT == "assistant"
def test_chat_role_is_string(self) -> None:
"""Test that ChatRole values are strings."""
assert isinstance(ChatRole.USER.value, str)
assert isinstance(ChatRole.SYSTEM.value, str)
assert isinstance(ChatRole.ASSISTANT.value, str)
class TestAgentSessionId:
"""Test suite for AgentSessionId."""
@@ -164,49 +149,55 @@ class TestRunRequest:
def test_init_with_defaults(self) -> None:
"""Test RunRequest initialization with defaults."""
request = RunRequest(message="Hello", conversation_id="conv-default")
request = RunRequest(message="Hello", thread_id="thread-default")
assert request.message == "Hello"
assert request.role == ChatRole.USER
assert request.role == Role.USER
assert request.response_format is None
assert request.enable_tool_calls is True
assert request.conversation_id == "conv-default"
assert request.thread_id == "thread-default"
def test_init_with_all_fields(self) -> None:
"""Test RunRequest initialization with all fields."""
schema = ModuleStructuredResponse
request = RunRequest(
message="Hello",
conversation_id="conv-123",
role=ChatRole.SYSTEM,
thread_id="thread-123",
role=Role.SYSTEM,
response_format=schema,
enable_tool_calls=False,
)
assert request.message == "Hello"
assert request.role == ChatRole.SYSTEM
assert request.role == Role.SYSTEM
assert request.response_format is schema
assert request.enable_tool_calls is False
assert request.conversation_id == "conv-123"
assert request.thread_id == "thread-123"
def test_init_coerces_string_role(self) -> None:
"""Ensure string role values are coerced into Role instances."""
request = RunRequest(message="Hello", thread_id="thread-str-role", role="system") # type: ignore[arg-type]
assert request.role == Role.SYSTEM
def test_to_dict_with_defaults(self) -> None:
"""Test to_dict with default values."""
request = RunRequest(message="Test message", conversation_id="conv-to-dict")
request = RunRequest(message="Test message", thread_id="thread-to-dict")
data = request.to_dict()
assert data["message"] == "Test message"
assert data["enable_tool_calls"] is True
assert data["role"] == "user"
assert "response_format" not in data or data["response_format"] is None
assert data["conversation_id"] == "conv-to-dict"
assert data["thread_id"] == "thread-to-dict"
def test_to_dict_with_all_fields(self) -> None:
"""Test to_dict with all fields."""
schema = ModuleStructuredResponse
request = RunRequest(
message="Hello",
conversation_id="conv-456",
role=ChatRole.ASSISTANT,
thread_id="thread-456",
role=Role.ASSISTANT,
response_format=schema,
enable_tool_calls=False,
)
@@ -218,17 +209,17 @@ class TestRunRequest:
assert data["response_format"]["module"] == schema.__module__
assert data["response_format"]["qualname"] == schema.__qualname__
assert data["enable_tool_calls"] is False
assert data["conversation_id"] == "conv-456"
assert data["thread_id"] == "thread-456"
def test_from_dict_with_defaults(self) -> None:
"""Test from_dict with minimal data."""
data = {"message": "Hello", "conversation_id": "conv-from-dict"}
data = {"message": "Hello", "thread_id": "thread-from-dict"}
request = RunRequest.from_dict(data)
assert request.message == "Hello"
assert request.role == ChatRole.USER
assert request.role == Role.USER
assert request.enable_tool_calls is True
assert request.conversation_id == "conv-from-dict"
assert request.thread_id == "thread-from-dict"
def test_from_dict_with_all_fields(self) -> None:
"""Test from_dict with all fields."""
@@ -241,38 +232,39 @@ class TestRunRequest:
"qualname": ModuleStructuredResponse.__qualname__,
},
"enable_tool_calls": False,
"conversation_id": "conv-789",
"thread_id": "thread-789",
}
request = RunRequest.from_dict(data)
assert request.message == "Test"
assert request.role == ChatRole.SYSTEM
assert request.role == Role.SYSTEM
assert request.response_format is ModuleStructuredResponse
assert request.enable_tool_calls is False
assert request.conversation_id == "conv-789"
assert request.thread_id == "thread-789"
def test_from_dict_invalid_role_defaults_to_user(self) -> None:
"""Test from_dict with invalid role defaults to USER."""
data = {"message": "Test", "role": "invalid_role", "conversation_id": "conv-invalid-role"}
def test_from_dict_with_unknown_role_preserves_value(self) -> None:
"""Test from_dict keeps custom roles intact."""
data = {"message": "Test", "role": "reviewer", "thread_id": "thread-with-custom-role"}
request = RunRequest.from_dict(data)
assert request.role == ChatRole.USER
assert request.role.value == "reviewer"
assert request.role != Role.USER
def test_from_dict_empty_message(self) -> None:
"""Test from_dict with empty message."""
data = {"conversation_id": "conv-empty"}
data = {"thread_id": "thread-empty"}
request = RunRequest.from_dict(data)
assert request.message == ""
assert request.role == ChatRole.USER
assert request.conversation_id == "conv-empty"
assert request.role == Role.USER
assert request.thread_id == "thread-empty"
def test_round_trip_dict_conversion(self) -> None:
"""Test round-trip to_dict and from_dict."""
original = RunRequest(
message="Test message",
conversation_id="conv-123",
role=ChatRole.SYSTEM,
thread_id="thread-123",
role=Role.SYSTEM,
response_format=ModuleStructuredResponse,
enable_tool_calls=False,
)
@@ -284,13 +276,13 @@ class TestRunRequest:
assert restored.role == original.role
assert restored.response_format is ModuleStructuredResponse
assert restored.enable_tool_calls == original.enable_tool_calls
assert restored.conversation_id == original.conversation_id
assert restored.thread_id == original.thread_id
def test_round_trip_with_pydantic_response_format(self) -> None:
"""Ensure Pydantic response formats serialize and deserialize properly."""
original = RunRequest(
message="Structured",
conversation_id="conv-pydantic",
thread_id="thread-pydantic",
response_format=ModuleStructuredResponse,
)
@@ -305,14 +297,14 @@ class TestRunRequest:
def test_init_with_correlation_id(self) -> None:
"""Test RunRequest initialization with correlation_id."""
request = RunRequest(message="Test message", conversation_id="conv-corr-init", correlation_id="corr-123")
request = RunRequest(message="Test message", thread_id="thread-corr-init", correlation_id="corr-123")
assert request.message == "Test message"
assert request.correlation_id == "corr-123"
def test_to_dict_with_correlation_id(self) -> None:
"""Test to_dict includes correlation_id."""
request = RunRequest(message="Test", conversation_id="conv-corr-to-dict", correlation_id="corr-456")
request = RunRequest(message="Test", thread_id="thread-corr-to-dict", correlation_id="corr-456")
data = request.to_dict()
assert data["message"] == "Test"
@@ -320,19 +312,19 @@ class TestRunRequest:
def test_from_dict_with_correlation_id(self) -> None:
"""Test from_dict with correlation_id."""
data = {"message": "Test", "correlation_id": "corr-789", "conversation_id": "conv-corr-from-dict"}
data = {"message": "Test", "correlation_id": "corr-789", "thread_id": "thread-corr-from-dict"}
request = RunRequest.from_dict(data)
assert request.message == "Test"
assert request.correlation_id == "corr-789"
assert request.conversation_id == "conv-corr-from-dict"
assert request.thread_id == "thread-corr-from-dict"
def test_round_trip_with_correlation_id(self) -> None:
"""Test round-trip to_dict and from_dict with correlation_id."""
original = RunRequest(
message="Test message",
conversation_id="conv-123",
role=ChatRole.SYSTEM,
thread_id="thread-123",
role=Role.SYSTEM,
correlation_id="corr-123",
)
@@ -342,7 +334,7 @@ class TestRunRequest:
assert restored.message == original.message
assert restored.role == original.role
assert restored.correlation_id == original.correlation_id
assert restored.conversation_id == original.conversation_id
assert restored.thread_id == original.thread_id
class TestAgentResponse:
@@ -351,12 +343,12 @@ class TestAgentResponse:
def test_init_with_required_fields(self) -> None:
"""Test AgentResponse initialization with required fields."""
response = AgentResponse(
response="Test response", message="Test message", conversation_id="conv-123", status="success"
response="Test response", message="Test message", thread_id="thread-123", status="success"
)
assert response.response == "Test response"
assert response.message == "Test message"
assert response.conversation_id == "conv-123"
assert response.thread_id == "thread-123"
assert response.status == "success"
assert response.message_count == 0
assert response.error is None
@@ -369,7 +361,7 @@ class TestAgentResponse:
response = AgentResponse(
response=None,
message="What is the answer?",
conversation_id="conv-456",
thread_id="thread-456",
status="success",
message_count=5,
error=None,
@@ -384,13 +376,13 @@ class TestAgentResponse:
def test_to_dict_with_text_response(self) -> None:
"""Test to_dict with text response."""
response = AgentResponse(
response="Text response", message="Message", conversation_id="conv-1", status="success", message_count=3
response="Text response", message="Message", thread_id="thread-1", status="success", message_count=3
)
data = response.to_dict()
assert data["response"] == "Text response"
assert data["message"] == "Message"
assert data["conversation_id"] == "conv-1"
assert data["thread_id"] == "thread-1"
assert data["status"] == "success"
assert data["message_count"] == 3
assert "structured_response" not in data
@@ -403,7 +395,7 @@ class TestAgentResponse:
response = AgentResponse(
response=None,
message="Question",
conversation_id="conv-2",
thread_id="thread-2",
status="success",
structured_response=structured,
)
@@ -417,7 +409,7 @@ class TestAgentResponse:
response = AgentResponse(
response=None,
message="Failed message",
conversation_id="conv-3",
thread_id="thread-3",
status="error",
error="Something went wrong",
error_type="ValueError",
@@ -434,7 +426,7 @@ class TestAgentResponse:
response = AgentResponse(
response="Text response",
message="Message",
conversation_id="conv-4",
thread_id="thread-4",
status="success",
structured_response=structured,
)
@@ -452,26 +444,26 @@ class TestModelIntegration:
def test_run_request_with_session_id(self) -> None:
"""Test using RunRequest with AgentSessionId."""
session_id = AgentSessionId.with_random_key("AgentEntity")
request = RunRequest(message="Test message", conversation_id=str(session_id))
request = RunRequest(message="Test message", thread_id=str(session_id))
assert request.conversation_id is not None
assert request.conversation_id == str(session_id)
assert request.conversation_id.startswith("@AgentEntity@")
assert request.thread_id is not None
assert request.thread_id == str(session_id)
assert request.thread_id.startswith("@AgentEntity@")
def test_response_from_run_request(self) -> None:
"""Test creating AgentResponse from RunRequest."""
request = RunRequest(message="What is 2+2?", conversation_id="conv-123", role=ChatRole.USER)
request = RunRequest(message="What is 2+2?", thread_id="thread-123", role=Role.USER)
response = AgentResponse(
response="4",
message=request.message,
conversation_id=request.conversation_id,
thread_id=request.thread_id,
status="success",
message_count=1,
)
assert response.message == request.message
assert response.conversation_id == request.conversation_id
assert response.thread_id == request.thread_id
if __name__ == "__main__":
python/packages/azurefunctions/tests/test_orchestration.py
+2 -2
View File
@@ -129,8 +129,8 @@ class TestDurableAIAgent:
assert request["enable_tool_calls"] is True
assert "correlation_id" in request
assert request["correlation_id"] == "correlation-guid"
assert "conversation_id" in request
assert request["conversation_id"] == "thread-guid"
assert "thread_id" in request
assert request["thread_id"] == "thread-guid"
def test_run_without_thread(self) -> None:
"""Test that run() works without explicit thread (creates unique session key)."""
python/samples/getting_started/azure_functions/01_single_agent/README.md
+9 -2
View File
@@ -7,7 +7,7 @@ This sample demonstrates how to use the Durable Extension for Agent Framework to
- Defining a simple agent with the Microsoft Agent Framework and wiring it into
an Azure Functions app via the Durable Extension for Agent Framework.
- Calling the agent through generated HTTP endpoints (`/api/agents/Joker/run`).
- Managing conversation state with session identifiers, so multiple clients can
- Managing conversation state with thread identifiers, so multiple clients can
interact with the agent concurrently without sharing context.
## Prerequisites
@@ -24,6 +24,13 @@ curl -X POST http://localhost:7071/api/agents/Joker/run \
-d "Tell me a short joke about cloud computing."
```
The agent responds with a JSON payload that includes the generated joke.
> **Note:** To return immediately with an HTTP 202 response instead of waiting for the agent output, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request body. The default behavior waits for the response.
## Expected Output
When you send a POST request with plain-text input, the Functions host responds with an HTTP 202 and queues the request for the durable agent entity. A typical response body looks like the following:
Expected HTTP 202 payload:
```json
@@ -31,7 +38,7 @@ Expected HTTP 202 payload:
"status": "accepted",
"response": "Agent request accepted",
"message": "Tell me a short joke about cloud computing.",
"conversation_id": "<guid>",
"thread_id": "<guid>",
"correlation_id": "<guid>"
}
```
python/samples/getting_started/azure_functions/01_single_agent/demo.http
+1 -3
View File
@@ -13,12 +13,10 @@ Content-Type: application/json
{
"message": "Add a security element to it.",
"sessionId": "session-003",
"waitForCompletion": true
"thread_id": "thread-001"
}
### Ask for a joke (plain text payload)
POST {{agentRoute}}/run
x-wait-for-completion: true
Give me a programming joke about race conditions.
python/samples/getting_started/azure_functions/02_multi_agent/README.md
+12 -3
View File
@@ -6,7 +6,7 @@ This sample demonstrates how to use the Durable Extension for Agent Framework to
- Using the Microsoft Agent Framework to define multiple AI agents with unique names and instructions.
- Registering multiple agents with the Function app and running them using HTTP.
- Conversation management (via session IDs) for isolated interactions per agent.
- Conversation management (via thread IDs) for isolated interactions per agent.
- Two different methods for registering agents: list-based initialization and incremental addition.
## Prerequisites
@@ -15,6 +15,15 @@ Complete the common environment preparation steps described in `../README.md`, i
## Running the Sample
With the environment setup and function app running, you can test the sample by sending HTTP requests to the different agent endpoints.
You can use the `demo.http` file to send messages to the agents, or a command line tool like `curl` as shown below:
> **Note:** Each endpoint waits for the agent response by default. To receive an immediate HTTP 202 instead, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request body.
### Test the Weather Agent
Bash (Linux/macOS/WSL):
Weather agent request:
```bash
@@ -30,7 +39,7 @@ Expected HTTP 202 payload:
"status": "accepted",
"response": "Agent request accepted",
"message": "What is the weather in Seattle?",
"conversation_id": "<guid>",
"thread_id": "<guid>",
"correlation_id": "<guid>"
}
```
@@ -50,7 +59,7 @@ Expected HTTP 202 payload:
"status": "accepted",
"response": "Agent request accepted",
"message": "Calculate a 20% tip on a $50 bill",
"conversation_id": "<guid>",
"thread_id": "<guid>",
"correlation_id": "<guid>"
}
```
python/samples/getting_started/azure_functions/02_multi_agent/demo.http
+2 -4
View File
@@ -36,8 +36,7 @@ Content-Type: application/json
{
"message": "What is the weather in Seattle?",
"sessionId": "weather-user-001",
"waitForCompletion": true
"thread_id": "weather-user-001"
}
###
@@ -51,8 +50,7 @@ Content-Type: application/json
{
"message": "Calculate a 20% tip on a $50 bill",
"sessionId": "math-user-001",
"waitForCompletion": true
"thread_id": "math-user-001"
}
###
python/samples/getting_started/azure_functions/03_callbacks/README.md
+7 -5
View File
@@ -10,8 +10,8 @@ an HTTP API that can be polled by a web client or dashboard.
- Registers a default `AgentResponseCallbackProtocol` implementation that logs streaming and final
responses.
- Persists callback events in an in-memory store and exposes them via
`GET /api/agents/{agentName}/callbacks/{conversationId}`.
- Shows how to reset stored callback events with `DELETE /api/agents/{agentName}/callbacks/{conversationId}`.
`GET /api/agents/{agentName}/callbacks/{thread_id}`.
- Shows how to reset stored callback events with `DELETE /api/agents/{agentName}/callbacks/{thread_id}`.
- Works alongside the standard `/api/agents/{agentName}/run` endpoint so you can correlate callback
telemetry with agent responses.
@@ -32,6 +32,8 @@ curl -X POST http://localhost:7071/api/agents/CallbackAgent/run \
-d '{"message": "Tell me a short joke"}'
```
> **Note:** The run endpoint waits for the agent response by default. To return immediately, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request body.
Poll callback telemetry (replace `<conversationId>` with the value from the POST response):
```bash
@@ -46,7 +48,7 @@ curl -X DELETE http://localhost:7071/api/agents/CallbackAgent/callbacks/<convers
## Expected Output
When you call `GET /api/agents/CallbackAgent/callbacks/{conversationId}` after sending a request to the agent,
When you call `GET /api/agents/CallbackAgent/callbacks/{thread_id}` after sending a request to the agent,
the API returns a list of streaming and final callback events similar to the following:
```json
@@ -54,7 +56,7 @@ the API returns a list of streaming and final callback events similar to the fol
{
"timestamp": "2024-01-01T00:00:00Z",
"agent_name": "CallbackAgent",
"conversation_id": "<conversationId>",
"thread_id": "<thread_id>",
"correlation_id": "<guid>",
"request_message": "Tell me a short joke",
"event_type": "stream",
@@ -64,7 +66,7 @@ the API returns a list of streaming and final callback events similar to the fol
{
"timestamp": "2024-01-01T00:00:01Z",
"agent_name": "CallbackAgent",
"conversation_id": "<conversationId>",
"thread_id": "<thread_id>",
"correlation_id": "<guid>",
"request_message": "Tell me a short joke",
"event_type": "final",
python/samples/getting_started/azure_functions/03_callbacks/demo.http
+7 -7
View File
@@ -3,13 +3,13 @@
###
### Endpoints introduced in this sample:
### - POST /api/agents/{agentName}/run : send a message to the agent
### - GET /api/agents/{agentName}/callbacks/{conversationId} : retrieve callback telemetry
### - DELETE /api/agents/{agentName}/callbacks/{conversationId} : clear stored callback events
### - GET /api/agents/{agentName}/callbacks/{thread_id} : retrieve callback telemetry
### - DELETE /api/agents/{agentName}/callbacks/{thread_id} : clear stored callback events
@baseUrl = http://localhost:7071
@agentName = CallbackAgent
@agentRoute = {{baseUrl}}/api/agents/{{agentName}}
@conversationId = test-stream-00
@thread_id = test-thread-00
### Health Check
GET {{baseUrl}}/api/health
@@ -20,11 +20,11 @@ Content-Type: application/json
{
"message": "Generate a short weather update for Paris and mention streaming callbacks.",
"sessionId": "{{conversationId}}"
"thread_id": "{{thread_id}}"
}
### Inspect callback telemetry
GET {{agentRoute}}/callbacks/{{conversationId}}
GET {{agentRoute}}/callbacks/{{thread_id}}
### Clear stored callback telemetry for the conversation
DELETE {{agentRoute}}/callbacks/{{conversationId}}
### Clear stored callback telemetry for the thread
DELETE {{agentRoute}}/callbacks/{{thread_id}}
python/samples/getting_started/azure_functions/03_callbacks/function_app.py
+40 -39
View File
@@ -24,7 +24,7 @@ from agent_framework.azurefunctions import AgentFunctionApp, AgentCallbackContex
logger = logging.getLogger(__name__)
# 1. Maintain an in-memory store for callback events (replace with durable storage in production).
# 1. Maintain an in-memory store for callback events keyed by thread ID (replace with durable storage in production).
CallbackStore = DefaultDict[str, list[dict[str, Any]]]
callback_events: CallbackStore = defaultdict(list)
@@ -65,8 +65,8 @@ class ConversationAuditTrail(AgentResponseCallbackProtocol):
"text": getattr(update, "text", None),
}
)
conversation_id = context.conversation_id or ""
callback_events[conversation_id].append(event)
thread_id = context.thread_id or ""
callback_events[thread_id].append(event)
preview = event.get("text") or event.get("update_kind")
self._logger.info(
@@ -85,8 +85,8 @@ class ConversationAuditTrail(AgentResponseCallbackProtocol):
"usage": _serialize_usage(getattr(response, "usage_details", None)),
}
)
conversation_id = context.conversation_id or ""
callback_events[conversation_id].append(event)
thread_id = context.thread_id or ""
callback_events[thread_id].append(event)
self._logger.info(
"[%s][%s] final response recorded",
@@ -96,10 +96,11 @@ class ConversationAuditTrail(AgentResponseCallbackProtocol):
@staticmethod
def _build_base_event(context: AgentCallbackContext) -> dict[str, Any]:
thread_id = context.thread_id
return {
"timestamp": datetime.now(timezone.utc).isoformat(),
"agent_name": context.agent_name,
"conversation_id": context.conversation_id,
"thread_id": thread_id,
"correlation_id": context.correlation_id,
"request_message": context.request_message,
}
@@ -122,12 +123,12 @@ app.add_agent(callback_agent)
@app.function_name("get_callback_events")
@app.route(route="agents/{agent_name}/callbacks/{conversationId}", methods=["GET"])
@app.route(route="agents/{agent_name}/callbacks/{thread_id}", methods=["GET"])
async def get_callback_events(req: func.HttpRequest) -> func.HttpResponse:
"""Return all callback events collected for a conversation."""
"""Return all callback events collected for a thread."""
conversation_id = req.route_params.get("conversationId", "")
events = callback_events.get(conversation_id, [])
thread_id = req.route_params.get("thread_id", "")
events = callback_events.get(thread_id, [])
return func.HttpResponse(
json.dumps(events, indent=2),
status_code=200,
@@ -136,44 +137,44 @@ async def get_callback_events(req: func.HttpRequest) -> func.HttpResponse:
@app.function_name("reset_callback_events")
@app.route(route="agents/{agent_name}/callbacks/{conversationId}", methods=["DELETE"])
@app.route(route="agents/{agent_name}/callbacks/{thread_id}", methods=["DELETE"])
async def reset_callback_events(req: func.HttpRequest) -> func.HttpResponse:
"""Clear the stored callback events for a conversation."""
"""Clear the stored callback events for a thread."""
conversation_id = req.route_params.get("conversationId", "")
callback_events.pop(conversation_id, None)
thread_id = req.route_params.get("thread_id", "")
callback_events.pop(thread_id, None)
return func.HttpResponse(status_code=204)
"""
Expected output when querying `GET /api/agents/CallbackAgent/callbacks/{conversationId}`:
Expected output when querying `GET /api/agents/CallbackAgent/callbacks/{thread_id}`:
HTTP/1.1 200 OK
[
{
"timestamp": "2024-01-01T00:00:00Z",
"agent_name": "CallbackAgent",
"conversation_id": "<conversationId>",
"correlation_id": "<guid>",
"request_message": "Tell me a short joke",
"event_type": "stream",
"update_kind": "text",
"text": "Sure, here's a joke..."
},
{
"timestamp": "2024-01-01T00:00:01Z",
"agent_name": "CallbackAgent",
"conversation_id": "<conversationId>",
"correlation_id": "<guid>",
"request_message": "Tell me a short joke",
"event_type": "final",
"response_text": "Why did the cloud...",
"usage": {
"type": "usage_details",
"input_token_count": 159,
"output_token_count": 29,
"total_token_count": 188
{
"timestamp": "2024-01-01T00:00:00Z",
"agent_name": "CallbackAgent",
"thread_id": "<thread_id>",
"correlation_id": "<guid>",
"request_message": "Tell me a short joke",
"event_type": "stream",
"update_kind": "text",
"text": "Sure, here's a joke..."
},
{
"timestamp": "2024-01-01T00:00:01Z",
"agent_name": "CallbackAgent",
"thread_id": "<thread_id>",
"correlation_id": "<guid>",
"request_message": "Tell me a short joke",
"event_type": "final",
"response_text": "Why did the cloud...",
"usage": {
"type": "usage_details",
"input_token_count": 159,
"output_token_count": 29,
"total_token_count": 188
}
}
}
]
"""
python/samples/getting_started/azure_functions/04_single_agent_orchestration_chaining/README.md
+2
View File
@@ -25,6 +25,8 @@ Poll the returned `statusQueryGetUri` until completion:
curl http://localhost:7071/api/singleagent/status/<instanceId>
```
> **Note:** The underlying agent run endpoint now waits for responses by default. If you invoke it directly and prefer an immediate HTTP 202, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the payload.
The orchestration first requests an inspirational sentence from the agent, then refines the initial response while
keeping it under 25 words—mirroring the behaviour of the corresponding .NET sample.
python/samples/getting_started/azure_functions/05_multi_agent_orchestration_concurrency/README.md
+2
View File
@@ -27,6 +27,8 @@ Poll the returned `statusQueryGetUri` until completion:
curl http://localhost:7071/api/multiagent/status/<instanceId>
```
> **Note:** The agent run endpoints wait for responses by default. If you call them directly and need an immediate HTTP 202, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request payload.
The orchestration launches both agents simultaneously so their domain-specific answers can be combined for the caller.
## Expected Output
python/samples/getting_started/azure_functions/06_multi_agent_orchestration_conditionals/README.md
+2
View File
@@ -27,6 +27,8 @@ Poll the returned `statusQueryGetUri` or call the status route directly:
curl http://localhost:7071/api/spamdetection/status/<instanceId>
```
> **Note:** The spam detection run endpoint waits for responses by default. To opt into an immediate HTTP 202, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the POST body.
## Expected Responses
- Spam payloads return `Email marked as spam: <reason>` by invoking the `handle_spam_email` activity.
- Legitimate emails return `Email sent: <draft>` after the email assistant agent produces a structured reply.
python/samples/getting_started/azure_functions/07_single_agent_orchestration_hitl/README.md
+2
View File
@@ -39,6 +39,8 @@ curl -X POST http://localhost:7071/api/hitl/approve/<instanceId> \
-d '{"approved": true, "feedback": "Looks good"}'
```
> **Note:** Calls to the underlying agent run endpoint wait for responses by default. If you need an immediate HTTP 202 response, set the `x-ms-wait-for-response` header or include `"wait_for_response": false` in the request body.
## Expected Responses
- `POST /api/hitl/run` returns a 202 Accepted payload with the Durable Functions instance ID.
- `POST /api/hitl/approve/{instanceId}` echoes the decision that the orchestration receives.