🧩

AIアプリケーションとJANコード付与ツールの結合の試み

catabon

2025/08/08に公開

FastAPI

Model Context Protocol

Agent

googleadk

tech

はじめに

WED株式会社でMLエンジニアをしている catabon です。WEDではユーザーの方々から買い取ったレシートに記載されている商品名に対して、データをより活用できるように、最も適切と思われるJANコードを検索し付与するシステムを日々運用しています。今回はこのシステムをより対話的に活用できないかと考えました。

FastAPI

まず、FastAPIを用いて、商品名が与えられたらそれに対応するJANコードを返すAPIを作成しました。これで、curlなどを用いてPOST方式でリクエストを送って、システムを実行することができるようになりました。

# Create FastAPI app
app = FastAPI(
    title="JAN Matching Engine API",
    description="API for matching product names to JAN codes",
    version="1.0.0"
)
...
@app.post("/api/match", operation_id="jan_search")
async def match_products(request):
...

Model Context Protocol（MCP）

次に、これをModel Context Protocol (MCP)サーバーとして, AIアプリケーションに組み合わせて実行できるようにしてみました。MCPはAIと外部サービスとの接続方法を標準化したもので、これによりAIアプリケーションに新たなサービスを追加しやすくなりました。更に、FastAPI-MCPというライブラリがあり、すでにFastAPIを用いてAPIを実装していれば、そのコードに対して以下のような行を追加するだけでMCPサーバーを立ちあげることができます。

from fastapi_mcp import FastApiMCP
...
mcp = FastApiMCP(app, name="MCP for JAN search")
mcp.mount()

これをAIアプリケーションで使いたいときには、設定ファイル(例えば Claude Desktopならclaude_desktop_config.json, Cursorならmcp.json)にmcpServersという項目が用意されているので、そこにMCPのアクセス先を以下のような形で追加すればOKです。

{
 "mcpServers": {
…
   "jan-mcp-server": {
     "command": "npx",
     "args": [
       "-y",
       "mcp-remote",
       "[tool_api_url]/mcp"
     ]
   },
…
 }
}

これで、AIアプリケーションにJANコードを調べるツールを追加できました。例えば以下のように、Claude Desktopを用いてまずWEB検索を行い、得られた複数の商品に対応するJANコードをこのツールで調べるといった使い方ができます。

"claude_mcp_screenshot"

Google Agent Development Kit (ADK)

このような形で複数のMCPを一つのAIアプリケーションに与えて、一連の処理を行わせることもできますが、実際に実行させようとするとプロンプトも長く複雑になり、また結果も安定しなくなってきます。できれば処理の組合せ方を事前に設定しておきたい、ということで、Google のエージェント開発キット（ADK）を使ってマルチエージェントで処理を実装してみました。

Googleの公式ブログにも紹介がありますが、Google ADKはGeminiを用いて、Web検索やBigQueryへのアクセスなどをエージェントとして組み合わせることができます。例えばGoogle検索はBuilt-inのツール(google_search)として用意されていて、以下のような形ですぐにエージェントとして用いることができます。

web_agent = Agent(
    name="basic_search_agent",
    model="gemini-2.0-flash",
    description="Agent to answer questions using Google Search.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    # google_search is a pre-built tool which allows the agent to perform Google searches.
    tools=[google_search]
)

BigQueryに対するアクセスも、SQLクエリを入力として受け取り検索結果をjsonとして出力するようなpythonの関数をツールとして用意して、それをエージェントから利用することができます。

# tools.py
def execute_query(query: str) -> str:
   try:
     client = bigquery.Client(project="[your_project_id]")
     job = client.query(query)
     result = job.result()
     result = [dict(row) for row in result]
     result = [{key: str(value) for key, value in raw.items()} for raw in result]
     return json.dumps(result, ensure_ascii=False)
   except Exception as e:
     print(f"Failed to run query: {e}")
     raise

エージェントの方は、instructionとしてテーブルの定義などを与え、toolsに先に定義した関数名を入れることでBigQueryの特定のテーブルにアクセスするように指定できます。

# agent.py
import os
from google.adk.agents import LlmAgent
from . import tools

instruction = """
[task]
Answer the question with the reason based on the data you get from BigQuery.

[condition]
Use SQL queries to get information from BigQuery
using the column definitions in the [table information].
...

[table information]
Columns of the table are as follows:
...
"""

bq_agent = LlmAgent(
   name="data_analysis_agent",
   model="gemini-2.0-flash",
   description="Agent to analysis bigquery data.",
   instruction=instruction,
   tools=[tools.execute_query],
)

既存のツール・サービスに加えて、自作のツールも同様の形で追加できます。例えばJANコードを付与するAPIを用いるエージェントも、ツールをpython関数として用意し、エージェントはそれをツールとして用いるように指示を与えることで作成できます。

# tools.py
import sys
import json
import requests

def search_product(chain_name: str, product_name: str) -> str:
    API_URL = "[jan_search_api_url]"
...
   result = {}
   try:
       response = requests.post(
           API_URL,
           json=input_data,
           headers={"Content-Type": "application/json"},
           timeout=10
       )
       result = response.json()
   except Exception as e:
       print(f"Error: {str(e)}")
       raise
   return result

# agent.py
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from . import tools
instruction = """
[task]
Search the product_name, and return the corresponding jan_code
and other obtained information using search_product_tool.
...
"""

search_product_tool = FunctionTool(func=tools.search_product)
jan_agent = LlmAgent(
   name="jan_matching_agent",
   model="gemini-2.0-flash",
   description="Search product with JAN Search.",
   instruction=instruction,
   tools=[search_product_tool],
)

個々のエージェントを接続する方法もまたエージェントの形で用意されています。エージェントを並列実行させるParallelAgentや、実行順序を指定するSequentialAgentを用いて、全体のワークフローを設計することができます。

search_parallel = ParallelAgent(
   name="ParallelSearchAgent",
   sub_agents=[bq_agent, jan_agent])

root_agent = SequentialAgent(
   name="SearchProductWorkflow",
   sub_agents=[web_agent, search_parallel, search_summary])