完成品のコードは以下のリポジトリにまとめています。

背景

まず、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を使用することが出来ました。

終わりに

以上で、自分なりのやり方を記載しました。
細かい実装内容などはもう少し改善できそうな気がするので、いい方法があればコメントしていただけると幸いです。