GitHub user wenjin272 created a discussion: Introduce YAML API
## 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`:
```yaml
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.
```yaml
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.
```yaml
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`.
```yaml
tools:
- name: my_tool
function: $module.my_tool
type: python
```
```yaml
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.
```yaml
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](https://claude.ai/chat/aa14ea1a-1959-4a33-a45e-3bb7a7783571#how-to-declare-and-reuse-common-resources-and-actions)](#how-to-declare-and-reuse-common-resources-and-actions)
below.
```yaml
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:
```yaml
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**
```yaml
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**
```yaml
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.
```python
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.
```python
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.
```json
{
"$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`:**
```python
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:**
```java
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.
GitHub link: https://github.com/apache/flink-agents/discussions/662
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
