AgnoからConnect AI経由でPingOne に接続する方法
Agno は、ツールを使って推論・計画・アクションを実行する AI エージェントを構築するための、開発者向け Python フレームワークです。 Agno はクリーンでコード駆動型のアーキテクチャを重視しており、エージェントランタイムを開発者が完全に制御できます。
CData Connect AI は、数百種類のエンタープライズデータソースを AI システムと連携させるための、セキュアなクラウド間インターフェースを提供します。 Connect AI を使用すると、リアルタイムの PingOne のデータ をレプリケーションなしでリモート MCP エンドポイント経由で公開できます。
このガイドでは、Agno Python SDK を使用して本番環境対応の Agno エージェントを構築します。 エージェントは streamable HTTP を使用して MCP 経由で CData Connect AI に接続し、利用可能なツールを動的に検出して、 リアルタイムの PingOne のデータ にクエリを実行します。
前提条件
- Python 3.9 以上
- CData Connect AI アカウント – こちらからサインアップまたはログイン
- 有効な認証情報を持つ PingOne アカウント
- LLM API キー（例：OpenAI）
概要
プロセスの概要は以下のとおりです。
- 接続：CData Connect AI で PingOne 接続を設定します。
- 検出：MCP を使用して CData Connect AI が公開するツールを動的に取得します。
- クエリ：MCP ツールを Agno 関数としてラップし、リアルタイムの PingOne のデータ にクエリを実行します。
Step 1: CData Connect AI で PingOne を設定
Agno からリアルタイムの PingOne のデータ にクエリを実行するには、まず CData Connect AI で PingOne 接続を作成します。 この接続は CData Remote MCP Server 経由で公開されます。
-
Connect AI にログインし、Sources をクリックしてから
Add Connection をクリックします。
-
Add Connection パネルから「PingOne」を選択します。
-
必要な認証プロパティを入力します。
PingOne に接続するには以下のプロパティを設定します。
- Region：自身のPingOne 組織のデータがホスティングされている地域。
- AuthScheme：PingOne に接続する際に使用する認証の種類。
- WorkerAppEnvironmentId （デフォルトのPingOne ドメインを使用する場合に必要）、またはAuthorizationServerURL のいずれかで、下で説明するように設定します。
WorkerAppEnvironmentId の設定
WorkerAppEnvironmentId は、Worker アプリケーションが存在するPingOne 環境のID です。 このパラメータは、環境がデフォルトのPingOne ドメイン（auth.pingone）を利用している場合のみ使用されます。 これは、ヘルプドキュメントのカスタムOAuth アプリケーションの作成で説明するように、PingOne への認証に使用するカスタムOAuth アプリケーションを作成した後に設定します。
はじめに、このプロパティの値を見つけます。
- 自身のPingOne 組織のホームページからナビゲーションサイドバーに移動し、Environments をクリックします。
- OAuth / Worker のカスタムアプリケーションを作成した環境（通常はAdministrators）を見つけ、Manage Environment をクリックします。 環境のホームページが表示されます。
- 環境のホームページのナビゲーションサイドバーで、Applications をクリックします。
- リストから、OAuth またはWorker アプリケーションの詳細を見つけます。
-
Environment ID フィールドの値をコピーします。
以下の例に似たものになるはずです：
WorkerAppEnvironmentId='11e96fc7-aa4d-4a60-8196-9acf91424eca'
次に、WorkerAppEnvironmentId をEnvironment ID フィールドの値に設定します。
AuthorizationServerURL の設定
AuthorizationServerURL は、お使いのアプリケーションが配置されている環境のPingOne 認可サーバーのベースURL です。 このプロパティは、PingOne プラットフォームAPI ドキュメントで説明されているように、環境にカスタムドメインを設定した場合にのみ使用されます。 Custom Domains を参照してください。
OAuth でのPingOne への認証
PingOne はOAuth とOAuthClient 認証の両方をサポートしています。 上述の設定手順に加え、OAuth またはOAuthCliet 認証をサポートするために、さらに2つの手順を完了する必要があります。
- ヘルプドキュメントのカスタムOAuth アプリケーションの作成で説明するように、カスタムOAuth アプリケーションを作成して設定します。
- ドライバーがデータモデル内のエンティティにアクセスできるようにするには、ヘルプドキュメントのAdministrator Roles での説明のとおり、使用するアドミンユーザー / ワーカーアプリケーションに対して正しいロールを設定していることを確認してください。
- 以下のサブセクションで説明されているように、選択した認証スキームと認証フローに適切なプロパティを設定します。
OAuth（認可コードグラント）
AuthScheme をOAuth に設定します。
デスクトップアプリケーション
OAuth アクセストークンの取得およびリフレッシュ
以下を設定して、接続してください。
- InitiateOAuth：GETANDREFRESH。繰り返しOAuth の交換を行ったり、手動でOAuthAccessToken を設定する必要をなくすには、InitiateOAuth を使用します。
- OAuthClientId：カスタムOAuth アプリケーションを作成した際に取得したClient ID。
- OAuthClientSecret：カスタムOAuth アプリケーションを作成した際に取得したClient Secret。
- CallbackURL：カスタムOAuth アプリケーションの登録時に定義したリダイレクトURI。例：https://localhost:3333
接続すると、CData 製品 はデフォルトブラウザでPingOne のOAuth エンドポイントを開きます。ログインして、アプリケーションにアクセス許可を与えます。 ドライバーはこれでOAuth プロセスを完了します。
- ドライバーはPingOne からアクセストークンを取得し、それを使ってデータをリクエストします。
- OAuth 値はOAuthSettingsLocation で指定された場所に保存され、接続間で永続化されるようにします。
ドライバーはアクセストークンの期限が切れると自動的にリフレッシュします。
Web アプリケーションやヘッドレスマシン、クライアントクレデンシャルグラントを含むその他のOAuth メソッドについては、ヘルプドキュメントを参照してください。
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 エージェントを作成してリアルタイムの PingOne のデータ にクエリ
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 ツールを呼び出し、リアルタイムの PingOne のデータ を返す様子を示しています。
これで、Agno エージェントを通じて自然言語でリアルタイムの PingOne のデータ にクエリを実行できるようになりました。
CData Connect AI の詳細
数百種類の SaaS、ビッグデータ、NoSQL ソースにクラウドアプリケーションから直接アクセスするには、 CData Connect AI をぜひお試しください。