<a id="flink-sql-create-agent"></a>

# CREATE AGENT Statement in Confluent Cloud

Confluent Intelligence enables creating Streaming Agents using declarative SQL syntax.
Agents can reason over streaming data, invoke tools, and produce structured
outputs.

## Syntax

### Standard agent

```sql
CREATE AGENT [IF NOT EXISTS] [catalog.][database.]agent_name
  USING MODEL <model_identifier>
  USING PROMPT <prompt_string>
  [USING TOOLS <tool1>, <tool2>]
  [COMMENT <comment_string>]
  WITH (<option_list>)
```

### Reflection agent

```sql
CREATE AGENT [IF NOT EXISTS] [catalog.][database.]agent_name
  USING AGENTS <agent1>, <agent2>
  [COMMENT <comment_string>]
  WITH (<option_list>)
```

## Description

Create a new Streaming Agent that can process streaming data using
LLM reasoning and tool invocation capabilities.

A standard agent uses a specified model, prompt, and optional tools
to process input data.

A reflection agent is a composite agent that orchestrates two
sub-agents (a drafter and a critic) in an iterative loop to produce
higher-quality output. The reflection agent does not have its own
model, prompt, or tools. For more information, see
[Improve Agent Output with Reflection Workflows](../../../ai/streaming-agents/reflection-workflow.md#streaming-agents-reflection-workflow).

## Parameters

- **model_identifier** (STRING): Reference to a registered model
  (standard agents only)
- **prompt_string** (STRING): System prompt for the agent (standard
  agents only)
- **tool1, tool2** (STRING): Comma-separated list of tool names
  (standard agents only)
- **agent1, agent2** (STRING): References to registered sub-agents
  (reflection agents only). The first agent is the drafter, and the
  second is the critic.
- **comment_string** (STRING): Optional comment describing the agent

## Agent options

### Standard agent options

- **max_consecutive_failures**: Maximum consecutive failures
  (optional, default: `3`)
- **max_iterations**: Maximum iterations for calling tools
  (optional, default: `20`)

### Reflection agent options

- **TYPE**: Agent type, must be `'REFLECTION'` (required)
- **PASS_CONDITION**: Termination condition for the reflection loop,
  `'APPROVAL'` or `'CONFIDENCE'` (optional, default:
  `APPROVAL`)
- **MAX_ITERATIONS**: Maximum reflection loop iterations (optional,
  default: `20`)
- **CONFIDENCE_VALUE**: Confidence threshold, 0.0-1.0, used when
  PASS_CONDITION is `'CONFIDENCE'` (optional, default: `0.8`)

## Examples

### Basic agent

```sql
CREATE AGENT weather_agent
  USING MODEL openai
  USING PROMPT 'Find weather info for provided city'
  USING TOOLS mcp_server, get_weather_tool
  WITH (
    'max_iterations' = '5'
  );
```

### Complete example

```sql
-- Create MCP server connection
CREATE CONNECTION mcp_connection
WITH (
  'type' = 'mcp_server',
  'api-key' = '<your-api-key>',
  'endpoint' = 'https://mcp.example.com'
);

-- Create tools
CREATE TOOL mcp_server
USING CONNECTION mcp_connection
WITH (
  'type' = 'mcp',
  'allowed_tools' = 'tool1,tool2',
  'request_timeout' = '30'
);

-- Register a function.
CREATE FUNCTION convert_to_celsius
USING JAR 'celsius.jar'
COMMENT 'function to convert degree to celsius';

-- Create a tool based on the function.
CREATE TOOL convert_to_celsius_tool
USING FUNCTION convert_to_celsius
WITH (
  'type' = 'function',
  'description' = 'This function can be used by model to convert degree to celsius'
);

-- 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'
);

-- Register a remote model.
CREATE MODEL openai
INPUT(text STRING)
OUTPUT (res STRING)
WITH (
  'provider' = 'openai',
  'task' = 'text_generation',
  'openai.model_version' = 'gpt-4o',
  'openai.connection' = 'openai_connection'
);

-- Create an agent that uses the model and tools.
CREATE AGENT weather_agent
USING MODEL openai
USING PROMPT 'Find weather info for provided city'
USING TOOLS mcp_server, convert_to_celsius_tool;
```

### Reflection agent example

```sql
-- Create the drafter agent.
CREATE AGENT claim_extractor
USING MODEL claims_model
USING PROMPT 'You are a claims intake agent. Extract a JSON
  object with fields: incident_type, damage_summary,
  estimated_severity (low/med/high), estimated_cost, location
  (lat/long), and any missing_fields.';

-- Create the critic agent.
CREATE AGENT claim_critic
USING MODEL claims_model
USING PROMPT 'You are a claims quality reviewer. Critique the
  extracted JSON for missing required fields, contradictions,
  or low-confidence assumptions.';

-- Create the reflection agent.
CREATE AGENT reflective_claim_intake
USING AGENTS claim_extractor, claim_critic
WITH (
  'type' = 'reflection',
  'pass_condition' = 'approval',
  'max_iterations' = '3'
);
```

### Constraints

The following constraints apply to reflection agents:

- You must not specify `USING MODEL`, `USING PROMPT`, or
  `USING TOOLS` on the reflection agent.
- You must specify exactly two sub-agents in the `USING AGENTS`
  clause. The first agent is the drafter, and the second is the
  critic.

## Related content

- [AI_RUN_AGENT Function](../functions/model-inference-functions.md#flink-sql-ai-run-agent-function)
- [CREATE MODEL Statement](create-model.md#flink-sql-create-model)
- [Improve Agent Output with Reflection Workflows](../../../ai/streaming-agents/reflection-workflow.md#streaming-agents-reflection-workflow)

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