
Agent Development KitでLine用エージェントを作成する
2025年05月21日
ADKLine
ClineでとりあえずLineエージェント動かすところまで作ったメモ。 一旦公式アカウントからメッセージ送る所までできた。
リファクタリング一切してない前提(他のコードコピーしていじってるので不要な部分が結構あり)。 コーディネータから呼び出される前提。
これは公式のMCP Serverが出たのでそのまま使わせてもらう。
事前準備
メッセージ送信用の公式アカウント作成やAPIのアクセストークンの取得や送信するユーザーID等は事前にしておく(Web検索すると色々出てくる)。
エージェントのプログラム
MCP Server含むエージェントのプログラム。
MCP ServerはDockerの方を使用。
# Add nest_asyncio to allow running asyncio.run within an existing event loop
import nest_asyncio
nest_asyncio.apply()
import os
import sys
import asyncio
from dotenv import load_dotenv
from google.genai import types
from google.adk.agents import Agent
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams, StdioServerParameters
from shared_libraries.error_handling import (
before_agent_callback,
after_agent_callback,
)
load_dotenv('.env')
parent_dir = os.path.dirname(os.path.abspath(__file__))
root_dir = os.path.dirname(parent_dir)
sys.path.append(parent_dir)
sys.path.append(root_dir)
from shared_libraries import constants
from . import prompt
AGENT_MODEL = constants.MODEL_GEMINI_2_5_PRO # Starting with a powerful Gemini model
# --- Helper function to initialize agent asynchronously ---
async def _initialize_agent():
"""Asynchronously fetches MCP tools and initializes the LlmAgent."""
print("Attempting to connect to MCP Line server and fetch tools...")
# Load connection details from environment variables
DESTINATION_USER_ID = constants.DESTINATION_USER_ID
CHANNEL_ACCESS_TOKEN = constants.CHANNEL_ACCESS_TOKEN
LINE_DOCKER_IMAGE_URL = constants.LINE_DOCKER_IMAGE_URL
tools, exit_stack = await MCPToolset.from_server(
connection_params=StdioServerParameters(
command='docker',
args=[
"run",
"-i",
"--rm",
"-e",
"CHANNEL_ACCESS_TOKEN",
"-e",
"DESTINATION_USER_ID",
LINE_DOCKER_IMAGE_URL
],
env={
"CHANNEL_ACCESS_TOKEN": CHANNEL_ACCESS_TOKEN,
"DESTINATION_USER_ID": DESTINATION_USER_ID
},
)
)
print(f"MCP Toolset created successfully. Fetched {len(tools)} tools.")
agent = Agent(
model=AGENT_MODEL, # Adjust model name if needed
name=prompt.AGENT_NAME,
description=prompt.DESCRIPTION,
instruction=prompt.PROMPT,
tools=tools,
before_agent_callback=before_agent_callback,
after_agent_callback=after_agent_callback,
)
print("Agent initialized.")
# Return both agent and exit_stack so cleanup can potentially be handled
return agent, exit_stack
# --- Top-level Agent Initialization ---
print("Initializing agent at module level...")
try:
# Run the async initialization function synchronously at the top level using nest_asyncio compatibility
line_root_agent, mcp_exit_stack = asyncio.run(_initialize_agent())
print("Agent initialization complete. 'root_agent' is now defined.")
# NOTE on Cleanup:
except Exception as e:
print(f"Error during agent initialization: {e}")
# Raise the exception to prevent the module from loading incorrectly
raise