<a id="streaming-agents-call-tools"></a>

# Call Tools in AI Workflows with Confluent Cloud

Streaming Agents can call external tools as part of AI workflows, enabling them
to access additional data sources, perform complex computations, or interact
with third-party services during agent execution. This capability enables
agents to invoke custom functions dynamically or connect to external APIs
and services, enhancing the flexibility and intelligence of your AI
applications.

Tools can be based on [user-defined functions (UDFs)](../../flink/concepts/user-defined-functions.md#flink-sql-udfs) or
Model Context Protocol (MCP) servers.

You can also create tools when you create an agent in the
Confluent Cloud Console. For more information, see
[Manage Streaming Agents in Confluent Cloud Console](manage-agents-console.md#streaming-agents-console).

You make a tool available to a streaming agent by using the
[CREATE TOOL](../../flink/reference/statements/create-tool.md#flink-sql-create-tool) statement to create a tool resource.
An agent can access multiple tools, guided by the agent’s prompt.

## Create and run an agent that calls tools

These are the general steps for creating and running a streaming agent that
calls tools.

- For a UDF-based tool:
  - Register a UDF by using the
    [CREATE FUNCTION](../../flink/reference/statements/create-function.md#flink-sql-create-function) statement.
  - Create a tool by using the [CREATE TOOL](../../flink/reference/statements/create-tool.md#flink-sql-create-tool)
    statement and referencing the function.
- For an MCP-based tool:
  - Create a connection to an MCP server by using the
    [CREATE CONNECTION](../../flink/reference/statements/create-connection.md#flink-sql-create-connection) statement.
  - Create a tool by using the [CREATE TOOL](../../flink/reference/statements/create-tool.md#flink-sql-create-tool)
    statement and referencing the connection.
- Create a remote model:
  - Create a connection to the model provider by using the
    [CREATE CONNECTION](../../flink/reference/statements/create-connection.md#flink-sql-create-connection) statement.
  - Register a remote model by using the
    [CREATE MODEL](../../flink/reference/statements/create-model.md#flink-sql-create-model) statement.
- Create a streaming agent that calls the tools by using the
  [CREATE AGENT](../../flink/reference/statements/create-agent.md#flink-sql-create-agent) statement.

#### NOTE
In reflection workflows, you define tools on the sub-agents
(drafter and critic), not on the composite reflection agent. For
more information, refer to
[Improve Agent Output with Reflection Workflows](reflection-workflow.md#streaming-agents-reflection-workflow).

- Run the agent on streaming data by using the
  [AI_RUN_AGENT](../../flink/reference/functions/model-inference-functions.md#flink-sql-ai-run-agent-function) function, which
  automates calls to the
  [AI_TOOL_INVOKE](../../flink/reference/functions/model-inference-functions.md#flink-sql-ai-tool-invoke-function)
  function.

## Call tools example

The following code example shows how to create a UDF-based tool and an
MCP-based tool.

```sql
-- Register a UDF.
CREATE FUNCTION udf_function
USING JAR 'udf.jar'
COMMENT 'UDF-based tool';

-- Create a tool based on the UDF.
CREATE TOOL udf_tool
USING FUNCTION udf_function
WITH (
  'type' = 'function',
  'description' = 'UDF-based tool'
);

-- Create a connection to an MCP server.
-- For streamable HTTP, you must set the transport type.
CREATE CONNECTION mcp_connection
WITH (
  'type' = 'mcp_server',
  'endpoint' = 'https://mcp.deepwiki.com/mcp',
  'api-key' = '<your-api-key>',
  'transport-type' = 'STREAMABLE_HTTP'
);

-- Create a tool based on the connection.
CREATE TOOL mcp_tool
USING CONNECTION mcp_connection
WITH (
  'type' = 'mcp',
  'allowed_tools' = 'read_wiki_structure,read_wiki_contents,ask_question'
);
```

The following code example shows how to create a remote model.

```sql
-- Create a connection to the model provider.
CREATE CONNECTION openai_connection
WITH (
  'type' = 'openai',
  'endpoint' = 'https://api.openai.com/v1/chat/completions',
  'api-key' = 'your-openai-key'
);

-- Create a remote model.
CREATE MODEL my_model
INPUT (message STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai',
  'openai.connection' = 'openai_connection',
  'openai.model_version' = 'gpt-4o',
  'task' = 'text_generation'
);
```

The following code example shows how to create a streaming agent that uses the
model and tools you created in the previous steps.

```sql
-- Create a streaming agent that uses the tools.
CREATE AGENT agent_with_tools
USING MODEL my_model
USING PROMPT 'You are a helpful agent that uses the tools to answer questions.'
USING TOOLS udf_tool, mcp_tool;
```

The following code example shows how to run the agent on streaming data by
using the [AI_RUN_AGENT](../../flink/reference/functions/model-inference-functions.md#flink-sql-ai-run-agent-function) function.

```sql
-- Create a table for prompts.
CREATE TABLE prompt_stream (
  id STRING,
  prompt STRING);

-- Insert a prompt into the table, with a generated ID.
INSERT INTO prompt_stream
  VALUES
    (UUID(), 'Fetch the content of https://www.amazon.com/primebigdealdays?ref_=nav_cs_td_pbdd_dt_cr and return price of any electronics mentioned in the page in Format {"Name": {"price", "discount"}} order by discount');

-- Run the agent on streaming data.
SELECT status, response FROM `prompt_stream`,
 LATERAL TABLE(AI_RUN_AGENT(`agent_with_tools`, `prompt`, `id`));
```

To monitor tool call activity during agent execution, you can enable
system table logging by passing a system log table name to
`AI_RUN_AGENT`. For more information, refer to
[Monitor Streaming Agents](monitor-streaming-agents.md#monitor-streaming-agents).

## Streaming Agent examples

Find more examples in the following topics:

- [Streaming Agents Examples](streaming-agents-examples.md#streaming-agents-examples)
- [Streaming Agents Quickstart repo](streaming-agents-examples.md#streaming-agents-examples-quick-start)

## Related content

- [Monitor Streaming Agents](monitor-streaming-agents.md#monitor-streaming-agents)
- [Agent Runtime Guide](agent-runtime-guide.md#streaming-agents-runtime)
- [Build and Run Streaming Agents](create-and-run-streaming-agents.md#streaming-agents-build-and-run)
- [Improve Agent Output with Reflection Workflows](reflection-workflow.md#streaming-agents-reflection-workflow)
- [Flink AI/ML Functions](../../flink/reference/functions/model-inference-functions.md#flink-sql-model-inference-functions)
- [Streaming Agents Overview](overview.md#ai-streaming-agents-overview)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).