Edit

Create and run dynamic workflows with Azure Functions hosted skills

In this article, you enable dynamic workflows for an Azure Functions hosted skill, create workflow-safe Python tools, and run a workflow locally. You can use the default Azure Storage backend or the recommended Durable Task Scheduler (DTS) backend.

Important

Azure Functions hosted skills is currently in preview. Features, configuration names, and supported connectors can change before general availability.

Prerequisites

Before you begin, you need:

You also need a storage backend for the workflow runtime. Choose the backend that fits your environment:

  • Azurite running locally, with AzureWebJobsStorage set to UseDevelopmentStorage=true in local.settings.json. You can instead use a connection string to an Azure Storage account.

No extra configuration is needed. The default Functions extension bundle includes the Durable Task extension.

For a complete project that demonstrates this workflow, see the workflow incident triage sample.

Enable workflows on the hosted skill

Add workflows.enabled: true to the front matter of the .agent.md file where you want to use dynamic workflows. The following example shows a hosted skill with workflows enabled:

---
name: Incident Triage Assistant
description: Investigates incidents by gathering evidence from multiple sources in parallel, correlating findings, and producing a written report.
builtin_endpoints: true
workflows:
  enabled: true
---

You are an incident-triage assistant. When a user describes a production incident, gather evidence from logs, metrics, and deployment history. When the work involves multiple evidence sources or a settling delay, run it as a workflow and summarize the final result for the user.

When you set workflows.enabled to true, the runtime adds the workflow-management tools described in the dynamic workflows overview.

Create workflow-safe tools

Dynamic workflows must only call tools that meet these requirements:

  • Run synchronously (not async).
  • Accept a single dict argument.
  • Return a JSON-serializable value.
  • Be idempotent, because a worker failure can cause the tool to run more than once.

Mark a function as workflow-safe by decorating it with @workflow_tool and placing it in the tools/ folder. If you also want the function available as a normal chat tool, add the @tool decorator alongside @workflow_tool.

The following example defines three workflow-safe tools. Two tools gather evidence, and one tool summarizes the results:

from typing import Any

from azure_functions_agents import workflow_tool


@workflow_tool(description="Fetch recent log lines for a service.")
def fetch_logs(args: dict[str, Any]) -> dict[str, Any]:
    service = args["service"]
    return {
        "service": service,
        "lines": [
            f"[ERROR] {service}: upstream timeout",
            f"[WARN] {service}: latency above SLO",
        ],
        "errors": 1,
        "warnings": 1,
    }


@workflow_tool(description="Fetch recent service metrics.")
def fetch_metrics(args: dict[str, Any]) -> dict[str, Any]:
    service = args["service"]
    return {
        "service": service,
        "cpu_p99": 86.2,
        "latency_p99_ms": 1420,
        "saturation": "high",
    }


@workflow_tool(description="Summarize logs and metrics into an incident finding.")
def summarize_findings(args: dict[str, Any]) -> dict[str, Any]:
    logs = args["logs"]
    metrics = args["metrics"]
    return {
        "service": logs["service"],
        "likely_cause": "resource pressure is correlated with elevated latency",
        "evidence": [
            f"{logs['errors']} error log entries found",
            f"p99 latency is {metrics['latency_p99_ms']} ms",
        ],
        "recommended_action": "scale out the service and review recent dependency timeouts",
    }

Understand the workflow plan

The model authors the plan that gets passed to start_workflow. The following simplified plan gathers logs and metrics in parallel, waits 30 seconds, and then summarizes the results:

{
  "tasks": [
    {
      "id": "logs",
      "type": "tool",
      "tool": "fetch_logs",
      "args": {
        "service": "orders-api"
      }
    },
    {
      "id": "metrics",
      "type": "tool",
      "tool": "fetch_metrics",
      "args": {
        "service": "orders-api"
      }
    },
    {
      "id": "settle",
      "type": "wait",
      "duration": "PT30S",
      "depends_on": [
        "logs",
        "metrics"
      ]
    },
    {
      "id": "summary",
      "type": "tool",
      "tool": "summarize_findings",
      "args": {
        "logs": "${logs.result}",
        "metrics": "${metrics.result}"
      },
      "depends_on": [
        "settle"
      ]
    }
  ]
}

Tip

You don't add this plan to the project as a static workflow definition. The model generates it at run time based on the hosted skill instructions, available workflow-safe tools, and the user's request.

In the generated plan, ${task_id.result} references the full JSON output from an upstream task. For example, ${logs.result} passes the entire return value of the logs task to summarize_findings. The model can also reference a specific field with ${task_id.result.path.to.field}.

Run locally

Start the Functions host and run a workflow by using the storage backend you chose in the prerequisites.

Azure Storage is the default backend. The default Functions extension bundle includes the Durable Task extension, so you don't need extra settings in host.json.

  1. Start Azurite.

  2. From the function app project root, start the Functions host.

    func start

  3. Open the built-in chat UI shown in the Core Tools output.

  4. Ask the hosted skill to perform work that uses multiple evidence sources. For example:

    Investigate latency spikes and intermittent 502 responses on orders-api.
Gather recent logs and metrics in parallel, wait 30 seconds, and summarize
the likely cause and recommended action.

The hosted skill starts a workflow and returns its workflow ID. The chat UI displays live progress and notifies the hosted skill when the workflow completes. The hosted skill then retrieves and summarizes the final result.

Related content

  • Last updated on