Introduce YAML API

[wenjin272]

Introduction

Flink-Agents currently provides Python and Java APIs. On top of these, we want to offer a declarative YAML API.

Compared with the programmatic APIs, the YAML API has the following advantages:

• Human-friendly. Low barrier to entry; agents can be templated and easily customized via parameter substitution.
• Coding-agent-friendly. A fixed schema saves tokens and enables strict schema validation. It also cleanly separates configuration changes (declared parameters) from logic changes (Action code).

Agent Declaration

This section describes how to declare agents with the YAML API, including:

• Declaring a single agent in a YAML file.
• Declaring multiple agents in a single YAML file.
• Declaring shared Resources and Actions alongside agents, and referencing them from the same file or from other YAML files.

How to declare a single agent

An agent typically consists of three kinds of components: Events, Resources, and Actions.

• Events: The unit of communication that flows inside an agent. Events are the signals that drive the whole workflow, and each event carries a type field that identifies its kind.
• Actions: Code snippets triggered by an event that emit new events. Actions hold the main business logic.
• Resources: Reusable, stateful components that actions can invoke at runtime — for example ChatModel, Tool, Prompt, and VectorStore.

Declaring an agent usually means declaring its Resources and Actions. The YAML below declares an agent named Example Agent:

agents:
  - name: Example Agent
    description: The example Agent

    # actions
    actions:
      - name: process_input
        function: flink_agents.examples.quickstart.agents.custom_types_and_resources.process_input
        type: python
        on: [input]
      - name: process_chat_response
        function: flink_agents.examples.quickstart.agents.custom_types_and_resources.process_chat_response
        type: python
        on: [chat_response]

    # resources
    chat_model_connections:
      - name: ollama_server
        clazz: ollama
        type: python
        base_url: http://localhost:11434

    chat_model_setups:
      - name: review_analysis_model
        clazz: ollama
        type: python
        connection: ollama_server
        model: qwen3:8b
        prompt: review_analysis_prompt
        tools: [notify_shipping_manager]
        extract_reasoning: true

    tools:
      - name: notify_shipping_manager
        function: flink_agents.examples.quickstart.agents.custom_types_and_resources.notify_shipping_manager
        type: python

Agent properties

• name (required): The agent's name.
• description (optional): A short description of what the agent does or how it works.
• namespace (optional): Corresponds to a Java package or a Python module.
  • When namespace is set on the agent, function may be omitted on Actions and Tools — the framework will derive the fully-qualified function path from namespace and name. This keeps YAML files more readable.

agents:
  - name: Example Agent
    namespace: flink_agents.examples.quickstart.agents.custom_types_and_resources

    tools:
      - name: notify_shipping_manager

    actions:
      - name: process_input
        type: python
        on: [input]

Resources

Prompt

• name (required): The prompt's name.
• messages / text (required): Correspond to Prompt.fromMessages and Prompt.fromText respectively. Exactly one of the two must be set.

prompts:
  - name: prompt1
    messages:
      - {role: system, content: "..."}
      - {role: user,   content: "{input}"}
  - name: prompt2
    text: "this is the {value}"

Tool

• name (required): The tool's name.
• function (optional):
  • The fully-qualified name of the method backing this function tool.
  • When the agent declares a namespace:
    • function may be just the method name; the framework will join namespace and function into a fully-qualified name.
    • function may also be omitted entirely; the tool's name will be treated as the method name, and the framework will join namespace and name into a fully-qualified name.
  • If both function and namespace are absent, an error is raised.
• type (optional): The implementation language of the function, java or python.

tools:
  - name: my_tool
    function: $module.my_tool
    type: python

agents:
  - name: my_agent
    namespace: $module

    tools:
      - name: my_tool

ResourceDescriptor-based Resources (chat model, vector store, embedding model, MCP, skills)

• name (required): The resource's name.
• clazz (required): The fully-qualified class name of the resource.
  • For built-in resources we will provide aliases — e.g. ollama, openai_completion, anthropic.
• type (optional): The implementation language of the resource, python or java.
• Additional keyword arguments (e.g. base_url, endpoint): Passed through verbatim to the resource implementation.

chat_model_connections:
  - name: my_connection
    clazz: ollama
    type: python
    base_url: http://localhost:11434

mcp_servers:
  - name: my_mcp
    clazz: mcp
    type: python
    endpoint: http://127.0.0.1:8000/mcp

Actions

actions is a list whose elements may each be either a map or a string.

• Map form:
  • name (required): The action's name.
  • function (optional): The fully-qualified name of the method backing this action. Same semantics as function on a tool.
  • type (optional): The implementation language of the action, python or java. Same semantics as type on a tool.
  • on (required): The event types this action listens to.
    • For built-in events we will provide aliases — e.g. input, chat_request, chat_response.
    • For custom events, the value must match the event's type string.
• String form: The name of a shared action defined elsewhere. See [How to declare and reuse common Resources and Actions](#how-to-declare-and-reuse-common-resources-and-actions) below.

actions:
  - name: action1
    function: $module.action1
    type: python
    on: [input]
  - name: action2
    function: $module.action2
    type: python
    on: [chat_response]
  - action3

How to declare multiple agents

Multiple agents may be declared in a single YAML file:

agents:
  - name: Agent1
    description: The first agent
    # resources
    # actions

  - name: Agent2
    description: The second agent
    # resources
    # actions

How to declare and reuse common Resources and Actions

Shared Resources and Actions can be declared at the same level as agents in a YAML file. Such shared components can be referenced by agents declared in the same file or in other YAML files.

Declaring and using shared Resources

agents:
  - name: Agent1
    description: The first agent
    chat_model_setups:
      - name: my_llm
        clazz: ollama
        connection: my_connection
        thinking: false

  - name: Agent2
    description: The second agent
    chat_model_setups:
      - name: my_llm
        clazz: ollama
        connection: my_connection
        thinking: true

# common resources for reuse
chat_model_connections:
  - name: my_connection
    clazz: ollama
    base_url: http://localhost:11434

Shared Resources are registered with the AgentsExecutionEnvironment. Any agent running in the same AgentsExecutionEnvironment can use them directly.

Note: Composing a single agent across multiple YAML files is not supported. If the YAML files parsed together contain duplicate agent names, or duplicate shared Resource or Action names, an exception is thrown.

Declaring and using shared Actions

agents:
  - name: Agent1
    description: The first agent

    actions:
      - action1
      - name: my_action
        function: $module.my_action
        on: [input]

  - name: Agent2
    description: The second agent

    actions:
      - action1

# common actions for reuse
actions:
  - name: action1
    function: $module.action1
    type: python
    on: [input]
  - name: action2
    function: $module.action2
    type: python
    on: [chat_response]

Agent Building

AgentsExecutionEnvironment exposes a load_yaml / loadYaml method that loads and parses one or more YAML files. The method:

• Registers shared Resources into the environment.
• Registers every declared agent into the environment.

class AgentsExecutionEnvironment:
    _resources: Dict[ResourceType, Dict[str, Any]]
    _agents: Dict[str, Agent]

    def load_yaml(self, paths: List[Path | str]) -> None:
        """Load and parse YAML files; all declared agents are registered into self._agents."""

• The method may be called multiple times. Results from successive calls are merged; an error is raised if a duplicate agent name, shared Resource, or shared Action is detected.
• Usage: callers can look up agents registered in the environment by name.

agents_env = AgentsExecutionEnvironment.get_execution_environment(env=env)
agents_env.load_yaml("path/to/yaml")

review_analysis_res_stream = (
    agents_env.from_datastream(
        input=product_review_stream, key_selector=lambda x: x.id
    )
    .apply("agent_name")
    .to_datastream()
)

YAML API Specification

To help users and coding agents understand and use the YAML API — and to make YAML files validatable — we provide a formal specification of the API. Because Flink-Agents exposes APIs in multiple languages, we describe the specification with JSON Schema, a language-neutral format.

The specification is codegen'd from the Python and Java data structures, so it never needs to be written by hand. This brings two benefits:

• A well-defined interface contract, easy for LLMs and other external systems to consume.
• Cross-language coverage, which allows us to verify API compatibility across language implementations.

JSON Schema

The JSON Schema below is illustrative — the final specification will continue to evolve alongside the YAML API.

{
  "$defs": {
    "ActionSpec": {
      "additionalProperties": false,
      "description": "An action references a user function and the event types it listens to.\n\nWhen ``function:`` is omitted, the loader falls back to\n``<namespace>.<name>``.",
      "properties": {
        "config": {
          "anyOf": [
            {
              "additionalProperties": true,
              "type": "object"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Config"
        },
        "function": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Function"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "on": {
          "items": {
            "type": "string"
          },
          "title": "On",
          "type": "array"
        }
      },
      "required": [
        "name",
        "on"
      ],
      "title": "ActionSpec",
      "type": "object"
    },
    "AgentSpec": {
      "additionalProperties": false,
      "description": "One agent inside a YAML file's ``agents:`` list.\n\nHolds the agent's own resources and actions. Resources/actions declared\nat the file level (siblings of ``agents:``) are merged in by the loader.",
      "properties": {
        "actions": {
          "items": {
            "anyOf": [
              {
                "$ref": "#/$defs/ActionSpec"
              },
              {
                "type": "string"
              }
            ]
          },
          "title": "Actions",
          "type": "array"
        },
        "chat_model_connections": {
          "items": {
            "$ref": "#/$defs/DescriptorSpec"
          },
          "title": "Chat Model Connections",
          "type": "array"
        },
        "chat_model_setups": {
          "items": {
            "$ref": "#/$defs/DescriptorSpec"
          },
          "title": "Chat Model Setups",
          "type": "array"
        },
        "description": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Description"
        },
        "embedding_model_connections": {
          "items": {
            "$ref": "#/$defs/DescriptorSpec"
          },
          "title": "Embedding Model Connections",
          "type": "array"
        },
        "embedding_model_setups": {
          "items": {
            "$ref": "#/$defs/DescriptorSpec"
          },
          "title": "Embedding Model Setups",
          "type": "array"
        },
        "events": {
          "items": {
            "$ref": "#/$defs/EventSchema"
          },
          "title": "Events",
          "type": "array"
        },
        "mcp_servers": {
          "items": {
            "$ref": "#/$defs/DescriptorSpec"
          },
          "title": "Mcp Servers",
          "type": "array"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "namespace": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Namespace"
        },
        "prompts": {
          "items": {
            "$ref": "#/$defs/PromptSpec"
          },
          "title": "Prompts",
          "type": "array"
        },
        "skills": {
          "items": {
            "$ref": "#/$defs/SkillsSpec"
          },
          "title": "Skills",
          "type": "array"
        },
        "tools": {
          "items": {
            "$ref": "#/$defs/ToolSpec"
          },
          "title": "Tools",
          "type": "array"
        },
        "vector_stores": {
          "items": {
            "$ref": "#/$defs/DescriptorSpec"
          },
          "title": "Vector Stores",
          "type": "array"
        }
      },
      "required": [
        "name"
      ],
      "title": "AgentSpec",
      "type": "object"
    },
    "DescriptorSpec": {
      "additionalProperties": true,
      "description": "Schema for any ResourceDescriptor-backed resource.\n\nRequired: ``name`` and ``clazz``. ``type`` selects the implementation\nlanguage (``\"python\"`` or ``\"java\"``; ``None`` means Python). All\nremaining fields are forwarded verbatim to ``ResourceDescriptor`` as\nkwargs (or as the Java wrapper's kwargs when ``type: java``).",
      "properties": {
        "clazz": {
          "title": "Clazz",
          "type": "string"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "type": {
          "anyOf": [
            {
              "enum": [
                "python",
                "java"
              ],
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Type"
        }
      },
      "required": [
        "name",
        "clazz"
      ],
      "title": "DescriptorSpec",
      "type": "object"
    },
    "EventAttribute": {
      "additionalProperties": false,
      "description": "Attribute spec under an event schema.",
      "properties": {
        "required": {
          "default": false,
          "title": "Required",
          "type": "boolean"
        },
        "type": {
          "title": "Type",
          "type": "string"
        }
      },
      "required": [
        "type"
      ],
      "title": "EventAttribute",
      "type": "object"
    },
    "EventSchema": {
      "additionalProperties": false,
      "description": "Advisory schema for a unified event's ``attributes`` map.",
      "properties": {
        "attributes": {
          "additionalProperties": {
            "$ref": "#/$defs/EventAttribute"
          },
          "title": "Attributes",
          "type": "object"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "strict": {
          "default": false,
          "title": "Strict",
          "type": "boolean"
        }
      },
      "required": [
        "name"
      ],
      "title": "EventSchema",
      "type": "object"
    },
    "MessageRole": {
      "description": "Role of a message in a chat conversation.",
      "enum": [
        "system",
        "user",
        "assistant",
        "tool"
      ],
      "title": "MessageRole",
      "type": "string"
    },
    "PromptMessage": {
      "additionalProperties": false,
      "description": "One message in a multi-turn prompt template.",
      "properties": {
        "content": {
          "title": "Content",
          "type": "string"
        },
        "role": {
          "$ref": "#/$defs/MessageRole",
          "default": "user"
        }
      },
      "required": [
        "content"
      ],
      "title": "PromptMessage",
      "type": "object"
    },
    "PromptSpec": {
      "additionalProperties": false,
      "description": "Declarative prompt: either a single ``text`` template or a list of\nrole-tagged ``messages``. Exactly one of the two fields must be set.",
      "properties": {
        "messages": {
          "anyOf": [
            {
              "items": {
                "$ref": "#/$defs/PromptMessage"
              },
              "type": "array"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Messages"
        },
        "name": {
          "title": "Name",
          "type": "string"
        },
        "text": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Text"
        }
      },
      "required": [
        "name"
      ],
      "title": "PromptSpec",
      "type": "object"
    },
    "SkillsSpec": {
      "additionalProperties": false,
      "description": "Declarative Skills resource pointing at one or more skill source\ndirectories on the local filesystem.",
      "properties": {
        "name": {
          "title": "Name",
          "type": "string"
        },
        "paths": {
          "items": {
            "type": "string"
          },
          "title": "Paths",
          "type": "array"
        }
      },
      "required": [
        "name",
        "paths"
      ],
      "title": "SkillsSpec",
      "type": "object"
    },
    "ToolSpec": {
      "additionalProperties": false,
      "description": "Points ``function:`` at a module attribute that is a callable tool.\n\nWhen ``function:`` is omitted, the loader falls back to\n``<namespace>.<name>``.",
      "properties": {
        "function": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Function"
        },
        "name": {
          "title": "Name",
          "type": "string"
        }
      },
      "required": [
        "name"
      ],
      "title": "ToolSpec",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Top-level YAML document.\n\nAlways wraps one or more agents under ``agents:``. Resources and\nactions declared at the same level as ``agents:`` are shared:\nresources are registered on the environment; actions can be\nreferenced from any agent by name string.",
  "properties": {
    "actions": {
      "items": {
        "$ref": "#/$defs/ActionSpec"
      },
      "title": "Actions",
      "type": "array"
    },
    "agents": {
      "items": {
        "$ref": "#/$defs/AgentSpec"
      },
      "title": "Agents",
      "type": "array"
    },
    "chat_model_connections": {
      "items": {
        "$ref": "#/$defs/DescriptorSpec"
      },
      "title": "Chat Model Connections",
      "type": "array"
    },
    "chat_model_setups": {
      "items": {
        "$ref": "#/$defs/DescriptorSpec"
      },
      "title": "Chat Model Setups",
      "type": "array"
    },
    "embedding_model_connections": {
      "items": {
        "$ref": "#/$defs/DescriptorSpec"
      },
      "title": "Embedding Model Connections",
      "type": "array"
    },
    "embedding_model_setups": {
      "items": {
        "$ref": "#/$defs/DescriptorSpec"
      },
      "title": "Embedding Model Setups",
      "type": "array"
    },
    "events": {
      "items": {
        "$ref": "#/$defs/EventSchema"
      },
      "title": "Events",
      "type": "array"
    },
    "mcp_servers": {
      "items": {
        "$ref": "#/$defs/DescriptorSpec"
      },
      "title": "Mcp Servers",
      "type": "array"
    },
    "prompts": {
      "items": {
        "$ref": "#/$defs/PromptSpec"
      },
      "title": "Prompts",
      "type": "array"
    },
    "skills": {
      "items": {
        "$ref": "#/$defs/SkillsSpec"
      },
      "title": "Skills",
      "type": "array"
    },
    "tools": {
      "items": {
        "$ref": "#/$defs/ToolSpec"
      },
      "title": "Tools",
      "type": "array"
    },
    "vector_stores": {
      "items": {
        "$ref": "#/$defs/DescriptorSpec"
      },
      "title": "Vector Stores",
      "type": "array"
    }
  },
  "required": [
    "agents"
  ],
  "title": "YamlAgentsDocument",
  "type": "object"
}

Mapping and validation in Python and Java

JSON Schema describes the YAML API specification in a language-neutral way. In Python it maps to a Pydantic BaseModel, and in Java to a POJO.

Python — Pydantic BaseModel:

class EventAttribute(BaseModel):
    """Attribute spec under an event schema."""

    model_config = ConfigDict(extra="forbid")

    type: str
    required: bool = False


class EventSchema(BaseModel):
    """Advisory schema for a unified event's ``attributes`` map."""

    model_config = ConfigDict(extra="forbid")

    name: str
    attributes: Dict[str, EventAttribute] = Field(default_factory=dict)
    strict: bool = False


class PromptMessage(BaseModel):
    """One message in a multi-turn prompt template."""

    model_config = ConfigDict(extra="forbid")

    role: MessageRole = MessageRole.USER
    content: str


class PromptSpec(BaseModel):
    """Declarative prompt: either a single ``text`` template or a list of
    role-tagged ``messages``.

    Exactly one of the two fields must be set.
    """

    model_config = ConfigDict(extra="forbid")

    name: str
    text: str | None = None
    messages: List[PromptMessage] | None = None

    @model_validator(mode="after")
    def _require_exactly_one(self) -> "PromptSpec":
        if (self.text is None) == (self.messages is None):
            msg = "prompt must define exactly one of 'text' or 'messages'"
            raise ValueError(msg)
        return self


class ToolSpec(BaseModel):
    """Points ``function:`` at a module attribute that is a callable tool.

    When ``function:`` is omitted, the loader falls back to
    ``<module>.<name>``.
    """

    model_config = ConfigDict(extra="forbid")

    name: str
    function: str | None = None


class DescriptorSpec(BaseModel):
    """Schema for any ResourceDescriptor-backed resource (chat model
    setups, chat model connections, embedding model setups, embedding
    model connections, vector stores, MCP servers).

    ``clazz`` is required and must be a fully-qualified class path; all
    other fields beyond ``name``/``clazz`` are passed through verbatim
    as ``ResourceDescriptor`` keyword arguments. Using ``extra="allow"``
    so callers can supply provider-specific options without needing a
    field for each one.
    """

    model_config = ConfigDict(extra="allow")

    name: str
    clazz: str

    def to_descriptor(self) -> ResourceDescriptor:
        """Build the ``ResourceDescriptor`` this spec describes."""
        kwargs = dict(self.model_extra or {})
        return ResourceDescriptor(clazz=self.clazz, **kwargs)


class ActionSpec(BaseModel):
    """An action references a user function and the event types it listens to.

    When ``function:`` is omitted, the loader falls back to
    ``<module>.<name>``.
    """

    model_config = ConfigDict(extra="forbid")

    name: str
    function: str | None = None
    on: List[str]
    config: Dict[str, Any] | None = None


class AgentDocument(BaseModel):
    """Top-level YAML document.

    Flat shape: ``name``/``description``/``module`` are top-level fields
    (no ``agent:`` wrapper); each resource category is its own top-level
    list (no ``resources:`` wrapper).
    """

    model_config = ConfigDict(extra="forbid")

    name: str | None = None
    description: str | None = None
    module: str | None = None

    events: List[EventSchema] = Field(default_factory=list)
    chat_model_connections: List[DescriptorSpec] = Field(default_factory=list)
    chat_model_setups: List[DescriptorSpec] = Field(default_factory=list)
    embedding_model_connections: List[DescriptorSpec] = Field(default_factory=list)
    embedding_model_setups: List[DescriptorSpec] = Field(default_factory=list)
    vector_stores: List[DescriptorSpec] = Field(default_factory=list)
    mcp_servers: List[DescriptorSpec] = Field(default_factory=list)
    prompts: List[PromptSpec] = Field(default_factory=list)
    tools: List[ToolSpec] = Field(default_factory=list)
    actions: List[ActionSpec] = Field(default_factory=list)

Java — POJO:

public class EventSchema {
    public String name;                          // required
    public Map<String, EventAttribute> attributes = new HashMap<>();
    public boolean strict = false;
}

public class EventAttribute {
    public String type;                          // required
    public boolean required = false;
}

public class PromptMessage {
    public MessageRole role = MessageRole.USER;
    public String content;                       // required
}

public class PromptSpec {
    public String name;                          // required
    public String text;                          // optional, mutually exclusive with messages
    public List<PromptMessage> messages;         // optional, mutually exclusive with text
}

public class ToolSpec {
    public String name;                          // required
    public String function;                      // optional, falls back to <module>.<name>
}

public class DescriptorSpec {
    // Use @JsonAnySetter / @JsonAnyGetter to accept provider-specific kwargs.
    public String name;                          // required
    public String clazz;                         // required
    @JsonIgnore public Map<String, Object> extras = new HashMap<>();
}

public class ActionSpec {
    public String name;                          // required
    public String function;                      // optional
    public List<String> on;                      // required, ≥1
    public Map<String, Object> config;           // optional
}

public class AgentDocument {
    public String name;
    public String description;
    public String module;
    public List<EventSchema> events = new ArrayList<>();
    public List<DescriptorSpec> chatModelConnections = new ArrayList<>();
    public List<DescriptorSpec> chatModelSetups = new ArrayList<>();
    public List<DescriptorSpec> embeddingModelConnections = new ArrayList<>();
    public List<DescriptorSpec> embeddingModelSetups = new ArrayList<>();
    public List<DescriptorSpec> vectorStores = new ArrayList<>();
    public List<DescriptorSpec> mcpServers = new ArrayList<>();
    public List<PromptSpec> prompts = new ArrayList<>();
    public List<ToolSpec> tools = new ArrayList<>();
    public List<ActionSpec> actions = new ArrayList<>();
}

When parsing a YAML file in Python or Java, Flink-Agents materializes it into the Pydantic BaseModel or Java POJO above, and relies on Pydantic / Jackson to perform schema validation.

Consistency guarantees

Because authoring a Pydantic BaseModel is considerably easier than hand-writing JSON Schema, the canonical JSON Schema file is exported from the Pydantic BaseModel and serves as the ground truth.

We will add tests to keep the Python API, Java API, and YAML API consistent:

• The JSON Schema exported from the Pydantic BaseModel matches the ground truth.
• The JSON Schema exported from the Java POJO matches the ground truth.
• The Pydantic BaseModel is consistent with the Python Agent API.
• The Java POJO is consistent with the Java Agent API.