Agno を使って CData Connect AI 経由で Zuora のデータと対話する
Agno は、ツールを使って推論・計画・アクションを実行する AI エージェントを構築するための、開発者向け Python フレームワークです。 Agno はクリーンでコード駆動型のアーキテクチャを重視しており、エージェントランタイムを開発者が完全に制御できます。
CData Connect AI は、300 以上のエンタープライズデータソースを AI システムと連携させるための、セキュアなクラウド間インターフェースを提供します。 Connect AI を使用すると、リアルタイムの Zuora のデータ をレプリケーションなしでリモート MCP エンドポイント経由で公開できます。
このガイドでは、Agno Python SDK を使用して本番環境対応の Agno エージェントを構築します。 エージェントは streamable HTTP を使用して MCP 経由で CData Connect AI に接続し、利用可能なツールを動的に検出して、 リアルタイムの Zuora のデータ にクエリを実行します。
前提条件
- Python 3.9 以上
- CData Connect AI アカウント – こちらからサインアップまたはログイン
- 有効な認証情報を持つ Zuora アカウント
- LLM API キー（例：OpenAI）
概要
プロセスの概要は以下のとおりです。
- 接続：CData Connect AI で Zuora 接続を設定します。
- 検出：MCP を使用して CData Connect AI が公開するツールを動的に取得します。
- クエリ：MCP ツールを Agno 関数としてラップし、リアルタイムの Zuora のデータ にクエリを実行します。
Step 1: CData Connect AI で Zuora を設定
Agno からリアルタイムの Zuora のデータ にクエリを実行するには、まず CData Connect AI で Zuora 接続を作成します。 この接続は CData Remote MCP Server 経由で公開されます。
-
Connect AI にログインし、Sources をクリックしてから
Add Connection をクリックします。
-
Add Connection パネルから「Zuora」を選択します。
-
必要な認証プロパティを入力します。
Zuora はユーザー認証にOAuth 標準を使用しています。OAuth 認証ついて詳しくは、オンラインヘルプドキュメントを参照してください。
Tenant プロパティの設定
プロバイダへの有効な接続を作成するには、アカウントの設定と合致するテナント値を1つ選択する必要があります。以下は、利用可能なオプションのリストです。- USProduction：リクエストはhttps://rest.zuora.com に送信されます。
- USAPISandbox：リクエストはhttps://rest.apisandbox.zuora.com に送信されます。
- USPerformanceTest：リクエストはhttps://rest.pt1.zuora.com に送信されます。
- EUProduction：リクエストはhttps://rest.eu.zuora.com に送信されます。
- EUSandbox：リクエストはhttps://rest.sandbox.eu.zuora.com に送信されます。
デフォルトではUSProduction テナントを使用します。
Zuora サービスの選択
データクエリとAQuA API の2つのZuora サービスを使用します。デフォルトでは、ZuoraService はAQuADataExport に設定されています。DataQuery
データクエリ機能は、非同期の読み取り専用SQL クエリを実行することで、Zuora テナントからのデータのエクスポートを実現します。 このサービスは、素早く軽量なSQL クエリでの使用を推奨します。制限
- フィルタ適用後の、テーブルごとの入力レコードの最大数： 1,000,000
- 出力レコードの最大数： 100,000
- テナントごとの、実行用に送信される同時クエリの最大数： 5
- テナントごとの、同時クエリの制限に達した後に実行用に送信され、キューに追加されるクエリの最大数： 10
- 1時間単位での、各クエリの最大処理時間： 1
- GB 単位での、各クエリに割り当てられるメモリの最大サイズ： 2
- Index Join を使用する際のインデックスの最大値。言い換えれば、Index Join を使用する際にWHERE 句で使われる一意の値に基づいた、左のテーブルから返されるレコードの最大数： 20.000
AQuADataExport
AQuA API のエクスポートは、すべてのオブジェクト（テーブル）のすべてのレコードをエクスポートするように設計されています。AQuA のクエリジョブには以下の制限があります。制限
- AQuA のジョブ内のクエリが8時間以上実行されている場合、ジョブは自動的に停止されます。
- 停止されたAQuA のジョブは3回再試行可能で、その後失敗として返されます。
Create & Test をクリックします。
-
Permissions タブを開き、ユーザーアクセスを設定します。
Personal Access Token の追加
Personal Access Token（PAT）は、Agno から CData Connect AI への MCP リクエストを認証するために使用されます。
- Settings を開き、Access Tokens に移動します。
- Create PAT をクリックします。
-
生成されたトークンを安全に保存します。
Step 2: 依存関係のインストールと環境変数の設定
Agno と MCP アダプターの依存関係をインストールします。LangChain は MCP ツールの互換性のためだけに含まれています。
pip install agno agno-mcp langchain-mcp-adapters
環境変数を設定します。
export CDATA_MCP_URL="https://mcp.cloud.cdata.com/mcp" export CDATA_MCP_AUTH="Base64EncodedCredentials" export OPENAI_API_KEY="your-openai-key"
「Base64EncodedCredentials」は、Connect AI ユーザーのメールアドレスと Personal Access Token をコロン（「:」）で結合し、Base64 エンコードした値です：Base64([email protected]:MY_CONNECT_AI_PAT)
Step 3: MCP 経由で CData Connect AI に接続
streamable HTTP を使用して MCP クライアントを作成します。これにより、CData Connect AI へのセキュアな接続が確立されます。
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
mcp_client = MultiServerMCPClient(
connections={
"default": {
"transport": "streamable_http",
"url": os.environ["CDATA_MCP_URL"],
"headers": {
"Authorization": f"Basic {os.environ['CDATA_MCP_AUTH']}"
}
}
}
)
Step 4: MCP ツールの検出
CData Connect AI は操作を MCP ツールとして公開します。これらは実行時に動的に取得されます。
langchain_tools = await mcp_client.get_tools() for tool in langchain_tools: print(tool.name)
Step 5: MCP ツールを Agno 関数に変換
各 MCP ツールを Agno 関数としてラップし、エージェントで使用できるようにします。
注意：Agno がすべての推論、計画、ツール選択を行います。LangChain は CData Connect AI が公開するツールを利用するための軽量な MCP 互換レイヤーとしてのみ使用されます。
from agno.tools import Function
def make_tool_caller(lc_tool):
async def call_tool(**kwargs):
return await lc_tool.ainvoke(kwargs)
return call_tool
Step 6: Agno エージェントを作成してリアルタイムの Zuora のデータ にクエリ
Agno がすべての推論、計画、ツール呼び出しを行います。LangChain は MCP 互換性以外の役割を果たしません。
from agno.agent import Agent
from agno.models.openai import OpenAIChat
agent = Agent(
model=OpenAIChat(
id="gpt-4o",
temperature=0.2,
api_key=os.environ["OPENAI_API_KEY"]
),
tools=agno_tools,
markdown=True
)
await agent.aprint_response(
"Show me the top 5 records from the available data source"
)
if __name__ == "__main__":
asyncio.run(main())
以下の結果は、Agno エージェントが CData Connect AI を通じて MCP ツールを呼び出し、リアルタイムの Zuora のデータ を返す様子を示しています。
これで、Agno エージェントを通じて自然言語でリアルタイムの Zuora のデータ にクエリを実行できるようになりました。
CData Connect AI の詳細
300 以上の SaaS、ビッグデータ、NoSQL ソースにクラウドアプリケーションから直接アクセスするには、 CData Connect AI をぜひお試しください。