Azure AI AgentとAzure SQL Databaseの連携方法を解説

2025.03.21

はじめに
前提条件
ライブラリ
ソースコード解説
まとめ

はじめに

MS開発部の松坂です。

Azure AI AgentとAzure SQL Databaseを連携させることで、データベースから取得した情報を活用したインテリジェントなアプリケーションを構築することができます。本記事では、Pythonを使った具体的なコード例を示しながら、その手順を解説します。

前提条件

以下の準備が整っていることを確認してください：

Azure SQL Database及びAzure AI Foundryが既に作成されていること。
実行環境でEntra ID認証が設定されていること。
Pythonコードが実行できる環境であること。

ライブラリ

pip install azure-ai-projects
pip install azure-identity
pip install pyodbc

pip install azure-ai-projects

pip install azure-identity

pip install pyodbc

ソースコード解説

1. Azure SQL Databaseへの接続設定

まず、Azure SQL Databaseに接続するための設定を行います。

database = ''
username = ''
password = ''
driver = '{ODBC Driver 17 for SQL Server}'
DATABASE_URI = f"DRIVER={driver};SERVER={database_server};PORT=1433;DATABASE={database_db};UID={database_user};PWD={database_password}"

database = ''

username = ''

password = ''

driver = '{ODBC Driver 17 for SQL Server}'

DATABASE_URI = f"DRIVER={driver};SERVER={database_server};PORT=1433;DATABASE={database_db};UID={database_user};PWD={database_password}"

2. SQLクエリ実行関数

次に、SQLクエリを実行して結果を取得する関数を定義します。

import pyodbc
import json
def execute_azure_sql_query(query: str) -&gt; str:
    try:
        with pyodbc.connect(DATABASE_URI) as connect:
            cursor = connect.cursor()
            cursor.execute(query)
            
            if query.strip().upper().startswith("SELECT"):
                columns = [column[0] for column in cursor.description]
                results = cursor.fetchall()
                data = []
                for row in results: 
                    if len(columns) != len(row):
                        raise ValueError(f"Mismatch between columns ({len(columns)}) and row ({len(row)}) lengths.")
                    
                    row_dict = {}
                    for column, value in zip(columns, row):
                        try:
                            row_dict[column] = value
                        except Exception as e:
                            print(f"Error serializing column: {column}, value: {value}, error: {str(e)}")
                            raise e
                    data.append(row_dict)
            else:
                connect.commit()
                data = {"message": "Query executed successfully"}
            
            return json.dumps(data, ensure_ascii=False)
    except Exception as e:
        print(str(e))
        return json.dumps({"error": str(e)})

import pyodbc

import json

def execute_azure_sql_query(query: str) -> str:

try:

with pyodbc.connect(DATABASE_URI) as connect:

cursor = connect.cursor()

cursor.execute(query)

if query.strip().upper().startswith("SELECT"):

columns = [column[0] for column in cursor.description]

results = cursor.fetchall()

data = []

for row in results:

if len(columns) != len(row):

raise ValueError(f"Mismatch between columns ({len(columns)}) and row ({len(row)}) lengths.")

row_dict = {}

for column, value in zip(columns, row):

try:

row_dict[column] = value

except Exception as e:

print(f"Error serializing column: {column}, value: {value}, error: {str(e)}")

raise e

data.append(row_dict)

else:

connect.commit()

data = {"message": "Query executed successfully"}

return json.dumps(data, ensure_ascii=False)

except Exception as e:

print(str(e))

return json.dumps({"error": str(e)})

ポイント解説

関数定義

- pyodbc.connect()でデータベースに接続します。

接続処理:
- pyodbc.connect()でデータベースに接続します。
- 接続文字列には先ほど定義した変数を利用します。
クエリ実行:
- cursor.execute(query)でSQLクエリ（例: "SELECT TOP 10 * FROM SalesOrderDetail"）を実行。
- 結果はcursor.fetchall()で取得し、リスト形式で返却します。
エラーハンドリング:
- 接続やクエリ実行中にエラーが発生した場合は例外処理でキャッチし、適切なメッセージを返却します。

3. Azure AI Agentの作成

次に、Azure AI Agentを作成します。

from azure.ai.projects import AIProjectClient
from azure.ai.projects.models import FunctionTool, ToolSet
from azure.identity import DefaultAzureCredential
project_client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(), conn_str=''
)
user_functions : Set[Callable[..., Any]] = {
    execute_azure_sql_query
}
toolset = ToolSet();
functions = FunctionTool(user_functions)
toolset.add(functions)
agent = project_client.agents.create_agent(
        model="gpt-4o-mini",
        name="my-agent",
        instructions="""
You are a database administrator. 
In future questions, first retrieve the relevant table information, 
then obtain the necessary column information, and only then perform any actions.
The questions may include aspects related to table structure, data retrieval, or updates. 
Please clearly explain how to proceed.
""",
        toolset=toolset,
)

from azure.ai.projects import AIProjectClient

from azure.ai.projects.models import FunctionTool, ToolSet

from azure.identity import DefaultAzureCredential

project_client = AIProjectClient.from_connection_string(

credential=DefaultAzureCredential(), conn_str=''

)

user_functions : Set[Callable[..., Any]] = {

execute_azure_sql_query

}

toolset = ToolSet();

functions = FunctionTool(user_functions)

toolset.add(functions)

agent = project_client.agents.create_agent(

model="gpt-4o-mini",

name="my-agent",

instructions="""

You are a database administrator.

In future questions, first retrieve the relevant table information,

then obtain the necessary column information, and only then perform any actions.

The questions may include aspects related to table structure, data retrieval, or updates.

Please clearly explain how to proceed.

""",

toolset=toolset,

)

ポイント解説

認証:
- DefaultAzureCredential()はAzure SDKで提供される認証方式で、環境変数やログイン情報から自動的に資格情報を取得します。
- conn_strにはAzure AI Foundryの接続文字列を設定してください。
AIエージェントの作成:
- create_agent()メソッドでAIエージェントを作成します。
- モデルには「GPT-4o」などの適切なモデル名を指定します。
- エージェントには「SQLデータベースを操作する」という指示（instructions）も与えます。
- 先ほど作成した関数をToolSetに設定します。

4. チャットの送受信

最後に、メッセージの送信と最後のメッセージ表示処理をまとめます。

thread = project_client.agents.create_thread()
 input_message = input("実行してほしい内容を書いてください：")
 message = project_client.agents.create_message(
        thread_id=thread.id,
        role="user",
        content=input_message
 )
 run = project_client.agents.create_and_process_run(thread_id=thread.id, assistant_id=agent.id)
 for data_point in reversed(messages.data):
        last_message_content = data_point.content[-1]
        role = data_point.role.value
        print(f"{role}: {last_message_content.text.value}")

thread = project_client.agents.create_thread()

input_message = input("実行してほしい内容を書いてください：")

message = project_client.agents.create_message(

thread_id=thread.id,

role="user",

content=input_message

)

run = project_client.agents.create_and_process_run(thread_id=thread.id, assistant_id=agent.id)

for data_point in reversed(messages.data):

last_message_content = data_point.content[-1]

role = data_point.role.value

print(f"{role}: {last_message_content.text.value}")

5. 結果

これらの処理をまとめて実行することで、DB名を指定しない、あいまいな内容のプロンプトでも、テーブルを特定して回答を得ることができました。

まとめ

本記事では、Azure AI AgentとAzure SQL Databaseの連携方法について、コード例と要所ごとの解説を行いました。この手順に従うことで、AIアプリケーションがデータベースと連携し、有益な情報提供が可能になります。また、SELECT文のようなデータ取得のほかにUPDATEやINSERTを実行するプロンプト指示も利用可能となります

以上、最後までご愛読いただき
ありがとうございました。

お問い合わせは、
以下のフォームへご連絡ください。

お問い合わせ