docs/feature/agent-tools.md

# Agent Tools & Event Triggers

For users who want an EventMesh-hosted agent to **act on** the mesh — deliver
LLM decisions through connectors, poll external systems mid-conversation, or
run agents fully event-driven without a human in the loop. Covers the
`eventmesh-agent` extension points (LLM client, conversation memory, tools,
event triggers) and how they map onto the connector plugin ecosystem.
*(Experimental)*

---

## The extension surface

`StreamingAgent` is assembled from constructor-injected interfaces — every
piece below has a shipped default and can be replaced without forking:

| Extension point | Interface | Default | Replace it to… |
| --- | --- | --- | --- |
| Chat backend | `agent.llm.LlmClient` | `OpenAiLlmClient` (OpenAI-compatible SSE) | call Anthropic native, Bedrock, or an internal inference service |
| Conversation history | `agent.ConversationMemory` | `ConversationStore` (in-memory window) | persist sessions to Redis / RocksDB / a database |
| Callable tools | `agent.tool.AgentTool` | none registered (plain streaming) | expose any capability to the model via function calling |
| Event triggers | `StreamingAgent.onEvent(...)` | off | drive the agent from topic subscriptions instead of user prompts |

## Connector plugins as agent tools

`ConnectorToolAdapter` turns the shipped connector SPI into `AgentTool`s, so
every connector plugin doubles as a tool the LLM may call:

- **Sink tool (write)** — the model's arguments object is wrapped as one
  CloudEvent and pushed through `SinkConnector.put()`. Example: a DingTalk
  sink connector becomes a `notify` tool the model can call to alert a
  channel.
- **Source tool (read)** — each call polls one batch via
  `SourceConnector.poll()` and returns it as a JSON array. Example: a canal
  source connector becomes a `query-changes` tool for mid-conversation data
  lookups.

Wire-up in `AgentApplication` (system properties):

```properties
# a write tool backed by any sink connector on the classpath
agent.tools.sink.notify=org.apache.eventmesh.connector.dingtalk.sink.DingTalkSinkConnector
agent.tools.props.notify.connector.sink.webhook=https://oapi.dingtalk.com/robot/send?...

# a read tool backed by any source connector
agent.tools.source.query-changes=org.apache.eventmesh.connector.canal.source.CanalSourceConnector
agent.tools.props.query-changes.connector.batchSize=10
```

With tools registered the agent switches from plain token streaming to a
**function-calling loop**: the model may request tools, receives their
results as messages, and continues until it produces a final answer (bounded
at 5 tool iterations). Without tools, behavior is unchanged
token-by-token streaming.

## In-tree agent plugins (`eventmesh-agent-plugin/`)

Beyond connector-backed tools, custom tools follow the repo-standard plugin
mechanism (same as storage/connector plugins):

1. Implement `agent.tool.AgentTool` (name + description + JSON schema +
   `invoke`), no extra annotation needed on your class.
2. In your jar add a service file
   `META-INF/eventmesh/org.apache.eventmesh.agent.tool.AgentTool` containing
   `mytool=com.example.MyTool`.
3. Drop the jar into the agent's `plugin/agent/` directory (the launcher
   puts every jar there on the classpath) and enable it with
   `AGENT_TOOLS_SPI=mytool` (i.e. `-Dagent.tools.spi=mytool`; comma-list
   supported).

Resolution goes through `EventMeshExtensionFactory` — the same loader that
serves storage and connector plugins, so singleton semantics and the
`META-INF/eventmesh/` convention are identical.

## In-tree agent plugins

The repo ships a plugin tree for agent tools — `eventmesh-agent-plugin/` — mirroring
`eventmesh-connector-plugin/`: each sub-module is a standalone jar with the
`META-INF/eventmesh/org.apache.eventmesh.agent.tool.AgentTool` service file, declared in
`gradle.properties` with `pluginType=agentTool` + `pluginName=<name>`. The `dist-agent` task
installs every agent plugin into `dist-agent/plugin/agent/<name>/` (the directory the agent
launcher puts on the classpath), so custom in-tree tools need zero wiring:

- `eventmesh-agent-plugin-http-fetch` — reference implementation: fetch a URL and return the
  (truncated) body to the model. Enable with `AGENT_TOOLS_SPI=http-fetch`.

Third-party jars follow the same shape: implement `AgentTool`, ship the service file, drop the
jar into `plugin/agent/`.

## Event-driven agents (no user in the loop)

Set a subscription list and an output topic:

```properties
agent.subscribe.topics=orders.changed,risk.alerts
agent.trigger.output.topic=agent.decisions
```

Each consumed CloudEvent becomes a prompt, is answered (with tools if
registered), and the answer is published onto the output topic — where any
**sink connector** can deliver it (DingTalk, Kafka, Spring, ...). Trigger
conversations are keyed `trigger:<eventId>` and never bleed into user
sessions. This is the agent-side twin of the connector pipeline:

```
source connector ──> topic ──> agent (trigger) ──> LLM (+ tools)
                                  │
                                  └── output topic ──> sink connector ──> external system
```

## Using the interfaces directly (embedder API)

```java
LlmClient llm = new OpenAiLlmClient(baseUrl, apiKey, model); // or your own impl
ConversationMemory memory = new ConversationStore(20);        // or a persistent impl
ToolRegistry tools = new ToolRegistry()
    .register(ConnectorToolAdapter.sinkTool("notify", "Send a DingTalk notification",
        DingTalkSinkConnector.class, props));
StreamingAgent agent = new StreamingAgent(client, parent, agentId, llm, memory, tools);
```

## Configuration reference

| Key | Default | Meaning |
| --- | --- | --- |
| `agent.tools.spi` | (empty) | Comma list of SPI names resolved via `EventMeshExtensionFactory` (jars in `plugin/agent/`) |
| `agent.tools.sink.<name>` | — | FQCN of a `SinkConnector` exposed as write tool `<name>` |
| `agent.tools.source.<name>` | — | FQCN of a `SourceConnector` exposed as read tool `<name>` |
| `agent.tools.props.<name>.*` | — | Connector init properties for tool `<name>` |
| `agent.subscribe.topics` | (empty) | Comma-separated topics consumed as event triggers |
| `agent.trigger.output.topic` | `agent.triggers` | Topic the trigger answers are published to |

The agent connects to the runtime like any SDK client (`agent.runtime.url`,
default `http://localhost:10105`).

## Where the code lives

| Piece | Location |
| --- | --- |
| LLM SPI + OpenAI default | `eventmesh-agent-runtime/.../agent/llm/{LlmClient,OpenAiLlmClient}.java` |
| Memory SPI + in-memory default | `eventmesh-agent-runtime/.../agent/{ConversationMemory,ConversationStore}.java` |
| Tool SPI + registry + adapter | `eventmesh-agent-runtime/.../agent/tool/{AgentTool,ToolRegistry,ConnectorToolAdapter}.java` |
| Tool loop + event trigger path | `eventmesh-agent-runtime/.../agent/StreamingAgent.java` |
| Boot wiring (tools + triggers) | `eventmesh-agent-runtime/.../agent/AgentApplication.java` |
| Tests | `eventmesh-agent-runtime/src/test/.../tool/*`, `.../llm/OpenAiLlmClientChatTest.java` |