完成品のコードは以下のリポジトリにまとめています。
背景
まず、Browser Useとは「ブラウザ操作を行えるAIエージェント」のことです。
私は以下の記事で知りました。
しかし、Browser Useで実行される関数は@controller.actionという独自のデコレータを使う必要があり、そのままではLangChainのtoolsを使うことはできません。
また、私はMCPサーバーで色々やっており、langchain-mcpという、LangChainのtoolsとしてそれを実行できるものがあるので、それを利用したいとも考えました。
そのため、一発で変換できるようにしてみました。
LangChainのtoolsを使う方法
まず、LangChainのtoolsを使うための方法です。
@controller.actionはどのようなものなのか、どこに登録されているかをコードから調べてみると、どうやらデコレータを付けた関数はRegisteredActionとして定義され、それをcontroller.registry.registry.actionsに追加されているようです。
それを元にtoolsを@controller.actionとして追加できるようにした関数は以下の通りです。
browser_use_tools.py
from browser_use.controller.service import Controller
from browser_use.controller.registry.views import RegisteredAction
class ToolFunc:
def __init__(self, tool):
self.tool = tool
async def call_langchain_tool(self, **kwargs):
ret = await self.tool.ainvoke(kwargs)
return ret
def add_langchain_tools_to_controller(langchain_tools: list, controller: Controller):
for langchain_tool in langchain_tools:
tool_func = ToolFunc(langchain_tool)
action = RegisteredAction(
name=langchain_tool.name,
description=langchain_tool.description,
function=tool_func.call_langchain_tool,
param_model=langchain_tool.args_schema,
requires_browser=False,
)
controller.registry.registry.actions[langchain_tool.name] = action
このadd_langchain_tools_to_controller()を以下のように使えば、toolsを@controller.actionとして使用できます。
index_tools.py
from typing import List
import asyncio
from pydantic import BaseModel
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from browser_use import Agent
from browser_use.controller.service import Controller
from browser_use_tools.browser_use_tools import add_langchain_tools_to_controller
controller = Controller()
class StoreResult(BaseModel):
price: str
postage: str
coupon: str
points: str
store: str
ecName: str
time: str
class StoreResults(BaseModel):
storeResults: List[StoreResult]
@tool(args_schema=StoreResults)
async def save_price(storeResults: List[StoreResult]):
"""Save Price"""
with open('price.txt', 'a', encoding='UTF-8') as f:
for storeResult in storeResults:
f.write(f'price:{storeResult.price},postage:{storeResult.postage},coupon:{storeResult.coupon},points:{storeResult.points},store:{storeResult.store},ecName:{storeResult.ecName},time:{storeResult.time}\n')
langchain_tools = [save_price, ]
add_langchain_tools_to_controller(langchain_tools, controller)
async def main():
agent = Agent(
task="""
あなたは価格監視のエージェントです。
与えられたURLから商品の監視をしてください
対象商品は: ロイヤルカナン 犬用 消化器サポート 低脂肪 小型犬用S 3kgx1
- Sundrug-online url: https://sundrug-online.com/products/3182550925792
下記の形式でデータを教えてほしい
- 価格
- 送料(なけれな0円)
- クーポン(なけれな0円)
- ポイント(なけれな0円)
- ショップ名
この結果はsave_priceで保存してください
""",
llm=ChatOpenAI(model="gpt-4o-mini"),
controller=controller
)
result = await agent.run()
print(result)
if __name__ == '__main__':
asyncio.run(main())
add_langchain_tools_to_controller(langchain_tools, controller)で、toolsのリストをControllerに追加しています。
なお、toolsの関数はasyncを付けたものだけ動作確認しています。
asyncの無い関数は確認していませんが、langchain-mcpはasyncを使うようになっており、また自作する場合でもasync付ければ特に問題ないはずです。
MCPサーバーのtoolsを使う方法
これでtoolsを使えるようになったので、先ほどのlangchain-mcpで、MCPサーバーのtoolsも使用できるようになりました。
langchain-mcpは基本的な使い方は公式のREADMEによれば以下の通りです。
demo.py
server_params = StdioServerParameters(
command="npx",
args=["-y", "@modelcontextprotocol/server-filesystem", str(pathlib.Path(__file__).parent.parent)],
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
toolkit = MCPToolkit(session=session)
await toolkit.initialize()
response = await run(toolkit.get_tools(), prompt)
print(response)
ただ、「複数のMCPサーバーを使うときに面倒」なのと「jsonファイルで設定出来るようにしたい」ということがありので、以下のようなクラスを自作して使っています。
mcp_manager.py
import os
import json
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp import MCPToolkit
class McpManager:
json_mtime = 0.0
mcp_tools = []
tasks = []
is_exit = False
loaded_tasks = 0
async def load(self, settings_file_path='settings/mcp_config.json'):
if os.path.isfile(settings_file_path) and self.json_mtime != os.path.getmtime(settings_file_path):
self.is_exit = True
await asyncio.gather(*self.tasks)
self.tasks = []
self.mcp_tools = []
self.is_exit = False
self.json_mtime = os.path.getmtime(settings_file_path)
with open(settings_file_path, mode='r', encoding='UTF-8') as f:
mcp_dict_all = json.load(f)
self.loaded_tasks = 0
for target in mcp_dict_all['mcpServers'].values():
self.tasks.append(asyncio.create_task(self.add_server(target)))
while self.loaded_tasks < len(self.tasks) and not self.is_exit:
await asyncio.sleep(0.1)
return True
return False
async def add_server(self, target):
if os.name == 'nt' and target['command'] != 'cmd':
server_params = StdioServerParameters(
command='cmd',
args=['/c', target['command'], *target['args']],
env=target['env'] if 'env' in target else None,
)
else:
server_params = StdioServerParameters(
command=target['command'],
args=target['args'],
env=target['env'] if 'env' in target else None,
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
toolkit = MCPToolkit(session=session)
await toolkit.initialize()
self.mcp_tools += toolkit.get_tools()
self.loaded_tasks += 1
while not self.is_exit:
await asyncio.sleep(0.1)
def get_tools(self):
return self.mcp_tools
def stop_servers(self):
self.is_exit = True
def __del__(self):
self.is_exit = True
これで「settings/mcp_config.json」に書かれたMCPサーバーを使うことが出来ます。
今回は「@modelcontextprotocol/server-filesystem」を使うので、「settings/mcp_config.json」は以下のようになります。
mcp_config.json
{
"mcpServers": {
"filesystem": {
"command": "npx",
"args": [
"-y",
"@modelcontextprotocol/server-filesystem",
"D:\\Python\\browser_use_tools\\Projects\\browser_use_tools"
]
}
}
}
実行するコードは以下の通りです。
index_mcp.py
import asyncio
from langchain_openai import ChatOpenAI
from browser_use import Agent
from browser_use.controller.service import Controller
from browser_use_tools.mcp_manager import McpManager
from browser_use_tools.browser_use_tools import add_langchain_tools_to_controller
controller = Controller()
async def main():
mcp_manager = McpManager()
await mcp_manager.load()
langchain_tools = mcp_manager.get_tools()
add_langchain_tools_to_controller(langchain_tools, controller)
agent = Agent(
task="""
test.txtをtestという内容で作成してください
ディレクトリの作成は不要です
""",
llm=ChatOpenAI(model="gpt-4o-mini"),
controller=controller
)
result = await agent.run()
print(result)
mcp_manager.stop_servers()
if __name__ == '__main__':
asyncio.run(main())
これでMCPサーバーのtoolsを使用することが出来ました。
終わりに
以上で、自分なりのやり方を記載しました。
細かい実装内容などはもう少し改善できそうな気がするので、いい方法があればコメントしていただけると幸いです。