Tool Use

Overview

Flink Agents provides a flexible and extensible tool use mechanism. Developers can define the tool as a local Python function, or they can integrate with a remote MCP server to use the tools provided by the MCP server.

Local Function as Tool

Developer can define the tool as a local Python/Java function, and there are two ways to define and register a local function as a tool:

Flink Agents uses the docstring of the python tool function to generate the tool metadata. The docstring of the python function should accurately describe the tool’s purpose, parameters, and return value, so that the LLM can understand the tool and use it effectively.

Define Tool as Static Method in Agent Class

Developer can define the tool as a static method in the agent class while defining the workflow agent, and use the @tool decorator to mark the function as a tool in python (or @Tool annotation in java). The tool can be referenced by its name in the tools list of the ResourceDescriptor when creating the chat model in the agent.

Python

class ReviewAnalysisAgent(Agent):

    @tool
    @staticmethod
    def notify_shipping_manager(id: str, review: str) -> None:
        """Notify the shipping manager when product received a negative review due to
        shipping damage.

        Parameters
        ----------
        id : str
            The id of the product that received a negative review due to shipping damage
        review: str
            The negative review content
        """
        notify_shipping_manager(id=id, review=review)

    @chat_model_setup
    @staticmethod
    def review_analysis_model() -> ResourceDescriptor:
        """ChatModel which focus on review analysis."""
        return ResourceDescriptor(
            clazz=ResourceName.ChatModel.OLLAMA_SETUP,
            ...,
            tools=["notify_shipping_manager"], # reference the tool by its name
        )

    ...

Java

public class ReviewAnalysisAgent extends Agent {

    @Tool(description = "Notify the shipping manager when product received a negative review due to shipping damage.")
    public static void notifyShippingManager(
            @ToolParam(name = "id") String id, @ToolParam(name = "review") String review) {
        CustomTypesAndResources.notifyShippingManager(id, review);
    }

    @ChatModelSetup
    public static ResourceDescriptor reviewAnalysisModel() {
        return ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
                .addInitialArgument("connection", "ollamaChatModelConnection")
                ...
                .addInitialArgument("tools", Collections.singletonList("notifyShippingManager")) // reference the tool by its name
                .build();
    }

    ...
}

Key points:

Use @tool decorator to define the tool in python (or @Tool annotation in java)
Reference the tool by its name in the tools list of the ResourceDescriptor

Register Tool to Execution Environment

Developer can register the tool to the execution environment, and then reference the tool by its name. This allows the tool to be reused by multiple agents.

Python

def notify_shipping_manager(id: str, review: str) -> None:
    """Notify the shipping manager when product received a negative review due to
    shipping damage.

    Parameters
    ----------
    id : str
        The id of the product that received a negative review due to shipping damage
    review: str
        The negative review content
    """
    ...

...

# Add notify shipping manager tool to the execution environment.
agents_env.add_resource(
    "notify_shipping_manager", ResourceType.TOOL, Tool.from_callable(notify_shipping_manager)
)

...

# Create react agent with notify shipping manager tool.
review_analysis_react_agent = ReActAgent(
    chat_model=ResourceDescriptor(
        clazz=ResourceName.ChatModel.OLLAMA_SETUP,
        tools=["notify_shipping_manager"], # reference the tool by its name
    ),
    ...
)

Java

@Tool(description = "Notify the shipping manager when product received a negative review due to shipping damage.")
public static void notifyShippingManager(
        @ToolParam(name = "id") String id, @ToolParam(name = "review") String review) {
    ...
}

// Add notify shipping manager tool to the execution environment.
agentsEnv
        .addResource(
                "notifyShippingManager",
                ResourceType.TOOL,
                org.apache.flink.agents.api.tools.Tool.fromMethod(
                        ReActAgentExample.class.getMethod(
                        "notifyShippingManager", String.class, String.class)));

// Create react agent with notify shipping manager tool.
ReActAgent reviewAnalysisReactAgent = new ReActAgent(
        ResourceDescriptor.Builder.newBuilder(ResourceName.ChatModel.OLLAMA_SETUP)
                .addInitialArgument(
                        "tools", Collections.singletonList("notifyShippingManager")) // reference the tool by its name
                ...
                .build(),
        ...);

Key points:

Use AgentsExecutionEnvironment.add_resource to register the tool to the execution environment
Reference the tool by its name in the tools list of the ResourceDescriptor

MCP Tool

See MCP for details.

Built-in Events and Actions

The built-in tool_call_action listens to ToolRequestEvent. For each tool call, it looks up the tool resource by function name, executes it through durable execution, and records whether it succeeded. After all tool calls in the batch have been processed, it sends a ToolResponseEvent.

When the tool request comes from chat_model_action, the emitted ToolResponseEvent is automatically consumed by chat_model_action to continue the chat. See Built-in Events and Actions in Chat Models for details on how chat_model_action handles tool responses.

Users can also send ToolRequestEvent directly when they want to invoke tools programmatically.