<a id="ai-invoke-tool-in-workflow"></a>

# Invoke a Tool in an AI Workflow with Confluent Cloud

The [AI_TOOL_INVOKE](../../flink/reference/functions/model-inference-functions.md#flink-sql-ai-tool-invoke-function) function in
Confluent Cloud for Apache Flink® enables dynamic tool invocation within an AI workflow. It enables
you to call external tools by using a Model Context Protocol (MCP) server or
local tools by using [user-defined functions (UDFs)](../../flink/concepts/user-defined-functions.md#flink-sql-udfs).

This capability is especially useful when a large language model (LLM)
determines that it needs to delegate part of its task to another system, like
performing a database lookup, calling a third-party API, or triggering an
operational action.

#### NOTE
The AI_TOOL_INVOKE function is available for preview.


A Preview feature is a Confluent Cloud component that is being introduced to gain
early feedback from developers. Preview features can be used for evaluation
and non-production testing purposes or to provide feedback to Confluent.
The warranty, SLA, and Support Services provisions of your agreement with
Confluent do not apply to Preview features. Confluent may discontinue
providing preview releases of the Preview features at any time in
Confluent’s’ sole discretion.

For help with the AI_TOOL_INVOKE function, contact
[streamingagents-early-access@confluent.io](mailto:streamingagents-early-access@confluent.io).

This guide shows how to use the AI_TOOL_INVOKE function to invoke tools
provided by an MCP server. You can also use the function to use local tools in
a UDF. The process is similar but requires you to create UDFs for the tools you
want to use.

- [Step 1: Create provider connections](#ai-invoke-tool-in-workflow-create-provider-connections): Create
  provider connections to the MCP server and the AI model.
- [Step 2: Create a model](#ai-invoke-tool-in-workflow-create-model): Create a model that uses the
  connections you created in the previous step.
- [Step 3: Run the AI model with the MCP tools](#ai-invoke-tool-in-workflow-run-model): Use the
  [AI_TOOL_INVOKE](../../flink/reference/functions/model-inference-functions.md#flink-sql-ai-tool-invoke-function) function in a SQL
  query to pass an input prompt to the model, along with descriptors for UDFs
  and tools you want to make available. The function returns a map containing
  the results from the tools, their responses, and the status of each call.

## Prerequisites

- Access to Confluent Cloud.
- Access to a Flink compute pool.
- Sufficient permissions to create models. For more information, see
  [RBAC for model inference](../confluent-intelligence/overview.md#ai-model-inference-rbac).
- A remote MCP server with an SSE endpoint. This guide uses the free MCP server
  at [https://mcp.deepwiki.com/sse](https://mcp.deepwiki.com/sse).
- A remote AI model. This guide assumes you have access to a model running on
  Azure OpenAI.

<a id="ai-invoke-tool-in-workflow-create-provider-connections"></a>

## Step 1: Create provider connections

In this step, you use Flink SQL to create connections to the MCP
server and the AI model. For more information, see
[CREATE CONNECTION](../../flink/reference/statements/create-connection.md#flink-sql-create-connection).

1. In the Flink SQL shell, or in a Flink workspace in Confluent Cloud Console, run
   the following [CREATE CONNECTION](../../flink/reference/statements/create-connection.md#flink-sql-create-connection) statement
   to create a connection to the MCP server.
   ```sql
   CREATE CONNECTION deepwiki_mcp_connection
     WITH (
       'type' = 'mcp_server',
       'endpoint' = 'https://mcp.deepwiki.com/sse',
       'api-key' = 'api_key'
     );
   ```
2. Run the following statement to create a connection to the AI model.
   ```sql
   # Create a connection to the AI model.
   CREATE CONNECTION azure-openai_connection
     WITH (
       'type' = 'azureopenai',
       'endpoint' = 'https://<your-model>.openai.azure.com/openai/deployments/<your-deployment-id>/chat/completions?api-version=2025-01-01-preview',
       'api-key' = '<your-api-key>'
     );
   ```

<a id="ai-invoke-tool-in-workflow-create-model"></a>

## Step 2: Create a model

In this step, you create a model that uses the AI model connection and the MCP
connection you created in the previous step.

1. In the Flink SQL shell, or in a Flink workspace in Confluent Cloud Console, run
   the following [CREATE MODEL](../../flink/reference/statements/create-model.md#flink-sql-create-model) statement to
   register your model.
   - The `system_prompt` property directs the remote model to select the best
     MCP tools to complete the task.
   - The `mcp.connection` property specifies the MCP server connection to
     use.

   ```sql
   CREATE MODEL `azureopenai_mcp_model`
   INPUT (prompt STRING)
   OUTPUT (response STRING)
   WITH (
     'provider' = 'azureopenai',
     'task' = 'text_generation',
     'azureopenai.system_prompt' = 'Use the best tool to respond to the input prompt',
     'azureopenai.connection' = 'azure-openai_connection',
     'mcp.connection' = 'deepwiki_mcp_connection'
   );
   ```
2. Run the following statement to create a table that contains the input
   prompts.
   ```sql
   CREATE TABLE text_stream (
     id BIGINT, prompt STRING
   );
   ```
3. Run the following statement to insert prompts into the table.
   ```sql
   INSERT INTO text_stream VALUES
     (1, 'What is the structure of the confluentinc/flink-cookbook repo?'),
     (2, 'Please get the docs for the confluentinc/flink-cookbook repo'),
     (3, 'ask_question tool: Does the confluentinc/flink-cookbook repo have examples for queries with tumbling windows?');
   ```

<a id="ai-invoke-tool-in-workflow-run-model"></a>

## Step 3: Run the AI model with the MCP tools

In this step, you run the AI model with the tools provided by the MCP
server.

1. Run the following statement to invoke your model.
   ```sql
   SELECT prompt, AI_TOOL_INVOKE(
     `azureopenai_mcp_model`,
     prompt,
     MAP[],
     MAP['read_wiki_structure', 'Get a list of documentation topics for a GitHub repository', 'read_wiki_contents', 'View documentation about a GitHub repository', 'ask_question', 'Ask any question about a GitHub repository']) AS response FROM text_stream;
   ```

   For the first prompt, your output should resemble:
   ```none
   read_wiki_structure,SUCCESS,Available pages for confluentinc/flink-cookbook:

   - 1 Flink Cookbook Overview
     - 1.1 Architecture and Build Patterns
     - 1.2 Getting Started
   - 2 Kafka Integration Recipes
     - 2.1 Dead Letter Queue Pattern
     - 2.2 Exactly-Once Processing
   ...
   ```

   For the second prompt, your output should resemble:
   ```none
   read_wiki_contents,SUCCESS,# Page: Flink Cookbook Overview

   # Flink Cookbook Overview

   <details>
   <summary>Relevant source files</summary>

   The following files were used as context for generating this wiki page:

   - [compiled-plan/pom.xml](compiled-plan/pom.xml)
   - [kafka-dead-letter/pom.xml](kafka-dead-letter/pom.xml)
   - [kafka-exactly-once/pom.xml](kafka-exactly-once/pom.xml)
   - [kafka-headers/pom.xml](kafka-headers/pom.xml)
   - [pattern-matching-cep/pom.xml](pattern-matching-cep/pom.xml)

   </details>

   This document provides an overview of the Flink Cookbook repository, which contains a collection of Apache Flink recipes demonstrating stream processing patterns, integrations, and best practices. Each recipe is implemented as a standalone Maven project with complete examples, tests, and documentation.

   ...
   ```

   For the third prompt, your output should resemble:
   ```none
   ask_question,SUCCESS,Yes, the repository `confluentinc/flink-cookbook` contains an example of a query using tumbling windows.  Specifically, the `kotlin` recipe demonstrates the use of `TumblingEventTimeWindows` for counting words within 5-second intervals.

   ### Tumbling Window Example

   The `WordCount.kt` file in the `kotlin` recipe defines a Flink application that processes text lines from a Kafka topic and counts words within tumbling windows.

   ...
   ```

## Related content

- [AI Model Inference Functions](../../flink/reference/functions/model-inference-functions.md#flink-sql-model-inference-functions)
- [Run an AI Model](../ai-model-inference.md#flink-sql-ai-model)

#### 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).