feat: add mcp sample

owl/run_mcp.py

@@ -0,0 +1,107 @@
+# run_mcp.py
+
+import asyncio
+import sys
+from pathlib import Path
+from typing import List
+
+from dotenv import load_dotenv
+
+from camel.models import ModelFactory
+from camel.toolkits import MCPToolkit, FunctionTool
+from camel.types import ModelPlatformType, ModelType
+from camel.logger import set_log_level
+
+from utils.async_role_playing import OwlRolePlaying, run_society
+
+from utils.mcp.mcp_toolkit_manager import MCPToolkitManager
+
+
+load_dotenv()
+set_log_level(level="DEBUG")
+
+
+async def construct_society(
+    question: str,
+    tools: List[FunctionTool],
+) -> OwlRolePlaying:
+    """
+    构建一个多Agent的OwlRolePlaying实例。
+    这里的tools已经是用户想交给assistant使用的全部Tool集合。
+    """
+    # 1. 创建模型
+    models = {
+        "user": ModelFactory.create(
+            model_platform=ModelPlatformType.OPENAI,
+            model_type=ModelType.GPT_4O,
+            model_config_dict={"temperature": 0},
+        ),
+        "assistant": ModelFactory.create(
+            model_platform=ModelPlatformType.OPENAI,
+            model_type=ModelType.GPT_4O,
+            model_config_dict={"temperature": 0},
+        ),
+    }
+
+    # 2. 配置User和Assistant
+    user_agent_kwargs = {"model": models["user"]}
+    assistant_agent_kwargs = {
+        "model": models["assistant"],
+        "tools": tools,  # 直接使用外部提供的全部tools
+    }
+
+    # 3. 设置任务参数
+    task_kwargs = {
+        "task_prompt": question,
+        "with_task_specify": False,
+    }
+
+    # 4. 构造并返回OwlRolePlaying
+    society = OwlRolePlaying(
+        **task_kwargs,
+        user_role_name="user",
+        user_agent_kwargs=user_agent_kwargs,
+        assistant_role_name="assistant",
+        assistant_agent_kwargs=assistant_agent_kwargs,
+    )
+    return society
+
+
+async def main():
+    # 准备MCP Servers
+    config_path = str(
+        Path(__file__).parent / "utils/mcp/mcp_servers_config.json"
+    )
+
+    manager = MCPToolkitManager.from_config(config_path)
+
+    # 示例问题
+    question = (
+        "I'd like a academic report about Guohao Li, including his research "
+        "direction, published papers (up to 20), institutions, etc." 
+        "Then organize the report in Markdown format and save it to my desktop"
+    )
+
+    # 在main中统一用async with把所有MCP连接打开
+    async with manager.connection():
+        # 这里 manager.is_connected() = True
+        # 获取合并后的tools
+        tools = manager.get_all_tools()
+
+        # 构造Society
+        society = await construct_society(question, tools)
+
+        # 运行对话
+        answer, chat_history, token_count = await run_society(society)
+
+    # 出了 with 块，这些toolkit就全部关闭
+    # manager.is_connected() = False
+
+    # 打印结果
+    print(f"\033[94mAnswer: {answer}\033[0m")
+    print("Chat History:", chat_history)
+    print("Token Count:", token_count)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

owl/utils/async_role_playing.py

@@ -0,0 +1,240 @@
+from typing import Optional, Tuple, List
+
+from camel.agents import ChatAgent
+from camel.responses import ChatAgentResponse
+from camel.messages.base import BaseMessage
+from camel.societies import RolePlaying
+from camel.logger import get_logger
+
+
+from copy import deepcopy
+
+logger = get_logger(__name__)
+
+
+class OwlRolePlaying(RolePlaying):
+    def __init__(self, **kwargs):
+        self.user_role_name = kwargs.get("user_role_name", "user")
+        self.assistant_role_name = kwargs.get("assistant_role_name", "assistant")
+
+        self.output_language = kwargs.get("output_language", None)
+
+        self.user_agent_kwargs = kwargs.get("user_agent_kwargs", {})
+        self.assistant_agent_kwargs = kwargs.get("assistant_agent_kwargs", {})
+
+        super().__init__(**kwargs)
+
+        init_user_sys_msg, init_assistant_sys_msg = self._construct_gaia_sys_msgs()
+
+        self.assistant_agent: ChatAgent
+        self.user_agent: ChatAgent
+        self.assistant_sys_msg: Optional[BaseMessage]
+        self.user_sys_msg: Optional[BaseMessage]
+
+        self._init_agents(
+            init_assistant_sys_msg,
+            init_user_sys_msg,
+            assistant_agent_kwargs=self.assistant_agent_kwargs,
+            user_agent_kwargs=self.user_agent_kwargs,
+            output_language=self.output_language,
+            # is_reasoning_task=self.is_reasoning_task
+        )
+
+    def _construct_gaia_sys_msgs(self):
+        user_system_prompt = f"""
+===== RULES OF USER =====
+Never forget you are a user and I am a assistant. Never flip roles! You will always instruct me. We share a common interest in collaborating to successfully complete a task.
+I must help you to complete a difficult task.
+You must instruct me based on my expertise and your needs to solve the task step by step. The format of your instruction is: `Instruction: [YOUR INSTRUCTION]`, where "Instruction" describes a sub-task or question.
+You must give me one instruction at a time.
+I must write a response that appropriately solves the requested instruction.
+You should instruct me not ask me questions.
+
+Please note that the task may be very complicated. Do not attempt to solve the task by single step. You must instruct me to find the answer step by step.
+Here are some tips that will help you to give more valuable instructions about our task to me:
+<tips>
+- I have various tools to use, such as search toolkit, web browser simulation toolkit, document relevant toolkit, code execution toolkit, etc. Thus, You must think how human will solve the task step-by-step, and give me instructions just like that. For example, one may first use google search to get some initial information and the target url, then retrieve the content of the url, or do some web browser interaction to find the answer.
+- Although the task is complex, the answer does exist. If you can’t find the answer using the current scheme, try to re-plan and use other ways to find the answer, e.g. using other tools or methods that can achieve similar results.
+- Always remind me to verify my final answer about the overall task. This work can be done by using multiple tools(e.g., screenshots, webpage analysis, etc.), or something else.
+- If I have written code, please remind me to run the code and get the result.
+- Search results typically do not provide precise answers. It is not likely to find the answer directly using search toolkit only, the search query should be concise and focuses on finding sources rather than direct answers, as it always need to use other tools to further process the url, e.g. interact with the webpage, extract webpage content, etc. 
+- If the question mentions youtube video, in most cases you have to process the content of the mentioned video.
+- For downloading files, you can either use the web browser simulation toolkit or write codes (for example, the github content can be downloaded via https://raw.githubusercontent.com/...).
+- Flexibly write codes to solve some problems, such as excel relevant tasks.
+</tips>
+
+Now, here is the overall task: <task>{self.task_prompt}</task>. Never forget our task!
+
+Now you must start to instruct me to solve the task step-by-step. Do not add anything else other than your instruction!
+Keep giving me instructions until you think the task is completed.
+When the task is completed, you must only reply with a single word <TASK_DONE>.
+Never say <TASK_DONE> unless my responses have solved your task.
+        """
+
+        assistant_system_prompt = f"""
+===== RULES OF ASSISTANT =====
+Never forget you are a assistant and I am a user. Never flip roles! Never instruct me! You have to utilize your available tools to solve the task I assigned.
+We share a common interest in collaborating to successfully complete a complex task.
+You must help me to complete the task.
+
+Here is our overall task: {self.task_prompt}. Never forget our task!
+
+I must instruct you based on your expertise and my needs to complete the task. An instruction is typically a sub-task or question.
+
+You must leverage your available tools, try your best to solve the problem, and explain your solutions.
+Unless I say the task is completed, you should always start with:
+Solution: [YOUR_SOLUTION]
+[YOUR_SOLUTION] should be specific, including detailed explanations and provide preferable detailed implementations and examples and lists for task-solving.
+
+Please note that our overall task may be very complicated. Here are some tips that may help you solve the task:
+<tips>
+- If one way fails to provide an answer, try other ways or methods. The answer does exists.
+- If the search snippet is unhelpful but the URL comes from an authoritative source, try visit the website for more details.  
+- When looking for specific numerical values (e.g., dollar amounts), prioritize reliable sources and avoid relying only on search snippets.  
+- When solving tasks that require web searches, check Wikipedia first before exploring other websites.  
+- When trying to solve math problems, you can try to write python code and use sympy library to solve the problem.
+- Always verify the accuracy of your final answers! Try cross-checking the answers by other ways. (e.g., screenshots, webpage analysis, etc.).  
+- Do not be overly confident in your own knowledge. Searching can provide a broader perspective and help validate existing knowledge.  
+- After writing codes, do not forget to run the code and get the result. If it encounters an error, try to debug it.
+- When a tool fails to run, or the code does not run correctly, never assume that it returns the correct result and continue to reason based on the assumption, because the assumed result cannot lead you to the correct answer. The right way is to think about the reason for the error and try again.
+- Search results typically do not provide precise answers. It is not likely to find the answer directly using search toolkit only, the search query should be concise and focuses on finding sources rather than direct answers, as it always need to use other tools to further process the url, e.g. interact with the webpage, extract webpage content, etc. 
+- For downloading files, you can either use the web browser simulation toolkit or write codes.
+</tips>
+
+        """
+
+        user_sys_msg = BaseMessage.make_user_message(
+            role_name=self.user_role_name, content=user_system_prompt
+        )
+
+        assistant_sys_msg = BaseMessage.make_assistant_message(
+            role_name=self.assistant_role_name, content=assistant_system_prompt
+        )
+
+        return user_sys_msg, assistant_sys_msg
+
+    async def astep(
+        self,
+        assistant_msg: BaseMessage
+    ) -> Tuple[ChatAgentResponse, ChatAgentResponse]:
+        user_response = await self.user_agent.astep(assistant_msg)
+        if user_response.terminated or user_response.msgs is None:
+            return (
+                ChatAgentResponse(msgs=[], terminated=False, info={}),
+                ChatAgentResponse(
+                    msgs=[],
+                    terminated=user_response.terminated,
+                    info=user_response.info,
+                ),
+            )
+        user_msg = self._reduce_message_options(user_response.msgs)
+
+        modified_user_msg = deepcopy(user_msg)
+
+        if "TASK_DONE" not in user_msg.content:
+            modified_user_msg.content += f"""\n
+            Here are auxiliary information about the overall task, which may help you understand the intent of the current task:
+            <auxiliary_information>
+            {self.task_prompt}
+            </auxiliary_information>
+            If there are available tools and you want to call them, never say 'I will ...', but first call the tool and reply based on tool call's result, and tell me which tool you have called.
+            """
+
+        else:
+            # The task is done, and the assistant agent need to give the final answer about the original task
+            modified_user_msg.content += f"""\n
+            Now please make a final answer of the original task based on our conversation : <task>{self.task_prompt}</task>
+            """
+
+        assistant_response = await self.assistant_agent.astep(user_msg)
+        if assistant_response.terminated or assistant_response.msgs is None:
+            return (
+                ChatAgentResponse(
+                    msgs=[],
+                    terminated=assistant_response.terminated,
+                    info=assistant_response.info,
+                ),
+                ChatAgentResponse(
+                    msgs=[user_msg], terminated=False, info=user_response.info
+                ),
+            )
+        assistant_msg = self._reduce_message_options(assistant_response.msgs)
+
+        modified_assistant_msg = deepcopy(assistant_msg)
+        if "TASK_DONE" not in user_msg.content:
+            modified_assistant_msg.content += f"""\n
+                Provide me with the next instruction and input (if needed) based on my response and our current task: <task>{self.task_prompt}</task>
+                Before producing the final answer, please check whether I have rechecked the final answer using different toolkit as much as possible. If not, please remind me to do that.
+                If I have written codes, remind me to run the codes.
+                If you think our task is done, reply with `TASK_DONE` to end our conversation.
+            """
+
+        return (
+            ChatAgentResponse(
+                msgs=[assistant_msg],
+                terminated=assistant_response.terminated,
+                info=assistant_response.info,
+            ),
+            ChatAgentResponse(
+                msgs=[user_msg],
+                terminated=user_response.terminated,
+                info=user_response.info,
+            ),
+        )
+
+
+async def run_society(
+    society: OwlRolePlaying,
+    round_limit: int = 15,
+) -> Tuple[str, List[dict], dict]:
+    overall_completion_token_count = 0
+    overall_prompt_token_count = 0
+
+    chat_history = []
+    init_prompt = """
+    Now please give me instructions to solve over overall task step by step. If the task requires some specific knowledge, please instruct me to use tools to complete the task.
+        """
+    input_msg = society.init_chat(init_prompt)
+    for _round in range(round_limit):
+        assistant_response, user_response = await society.astep(input_msg)
+        overall_prompt_token_count += (
+            assistant_response.info["usage"]["completion_tokens"]
+        )
+        overall_prompt_token_count += (
+            assistant_response.info["usage"]["prompt_tokens"]
+            + user_response.info["usage"]["prompt_tokens"]
+        )
+
+        # convert tool call to dict
+        tool_call_records: List[dict] = []
+        for tool_call in assistant_response.info["tool_calls"]:
+            tool_call_records.append(tool_call.as_dict())
+
+        _data = {
+            "user": user_response.msg.content,
+            "assistant": assistant_response.msg.content,
+            "tool_calls": tool_call_records,
+        }
+
+        chat_history.append(_data)
+        logger.info(f"Round #{_round} user_response:\n {user_response.msgs[0].content}")
+        logger.info(
+            f"Round #{_round} assistant_response:\n {assistant_response.msgs[0].content}"
+        )
+
+        if (
+            assistant_response.terminated
+            or user_response.terminated
+            or "TASK_DONE" in user_response.msg.content
+        ):
+            break
+
+        input_msg = assistant_response.msg
+
+    answer = chat_history[-1]["assistant"]
+    token_info = {
+        "completion_token_count": overall_completion_token_count,
+        "prompt_token_count": overall_prompt_token_count,
+    }
+
+    return answer, chat_history, token_info

owl/utils/mcp/mcp_servers_config.json

@@ -0,0 +1,20 @@
+{
+  "mcpServers": {
+    "filesystem": {
+      "command": "mcp-filesystem-server",
+      "args": [
+        "/Users/coco/Desktop",
+        "/Users/coco/Downloads"
+      ]
+    },
+    "simple-arxiv": {
+      "command": "python",
+      "args": ["-m", "mcp_simple_arxiv"]
+    }
+  },
+  "mcpWebServers": {
+    "weather": {
+      "url": "https://c9a9-89-185-25-132.ngrok-free.app/sse"
+    }
+  }
+}

owl/utils/mcp/mcp_toolkit_manager.py

@@ -0,0 +1,77 @@
+import json
+import os
+from typing import List, Optional, AsyncGenerator
+
+from camel.toolkits import MCPToolkit
+from contextlib import AsyncExitStack, asynccontextmanager
+
+
+class MCPToolkitManager:
+    """
+    负责管理多个 MCPToolkit 实例，并提供统一的连接管理。
+    """
+
+    def __init__(self, toolkits: List[MCPToolkit]):
+        self.toolkits = toolkits
+        self._exit_stack: Optional[AsyncExitStack] = None
+        self._connected = False
+
+
+    @staticmethod
+    def from_config(config_path: str) -> "MCPToolkitManager":
+        """从 JSON 配置文件加载 MCPToolkit 实例，并返回 MCPToolkitManager 实例。
+
+        :param config_path: JSON 配置文件路径
+        :return: MCPToolkitManager 实例
+        """
+        with open(config_path, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        all_toolkits = []
+
+        # 处理本地 MCP 服务器
+        mcp_servers = data.get("mcpServers", {})
+        for name, cfg in mcp_servers.items():
+            toolkit = MCPToolkit(
+                command_or_url=cfg["command"],
+                args=cfg.get("args", []),
+                env={**os.environ, **cfg.get("env", {})},
+                timeout=cfg.get("timeout", None),
+            )
+            all_toolkits.append(toolkit)
+
+        # 处理远程 MCP Web 服务器
+        mcp_web_servers = data.get("mcpWebServers", {})
+        for name, cfg in mcp_web_servers.items():
+            toolkit = MCPToolkit(
+                command_or_url=cfg["url"],
+                timeout=cfg.get("timeout", None),
+            )
+            all_toolkits.append(toolkit)
+
+        return MCPToolkitManager(all_toolkits)
+
+    @asynccontextmanager
+    async def connection(self) -> AsyncGenerator["MCPToolkitManager", None]:
+        """统一打开多个 MCPToolkit 的连接，并在离开上下文时关闭。"""
+        self._exit_stack = AsyncExitStack()
+        try:
+            # 顺序进入每个 toolkit 的 async context
+            for tk in self.toolkits:
+                await self._exit_stack.enter_async_context(tk.connection())
+            self._connected = True
+            yield self
+        finally:
+            self._connected = False
+            await self._exit_stack.aclose()
+            self._exit_stack = None
+
+    def is_connected(self) -> bool:
+        return self._connected
+
+    def get_all_tools(self):
+        """合并所有 MCPToolkit 提供的工具"""
+        all_tools = []
+        for tk in self.toolkits:
+            all_tools.extend(tk.get_tools())
+        return all_tools

owl/utils/mcp/servers/mcp_server.py

@@ -0,0 +1,136 @@
+# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ========= Copyright 2023-2024 @ CAMEL-AI.org. All Rights Reserved. =========
+from typing import Any
+
+import httpx
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP("weather")
+
+NWS_API_BASE = "https://api.weather.gov"
+USER_AGENT = "weather-app/1.0"
+
+
+async def make_nws_request(url: str) -> dict[str, Any] | None:
+    r"""Make a request to the NWS API with proper error handling."""
+    headers = {"User-Agent": USER_AGENT, "Accept": "application/geo+json"}
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.get(url, headers=headers, timeout=30.0)
+            response.raise_for_status()
+            return response.json()
+        except Exception:
+            return None
+
+
+def format_alert(feature: dict) -> str:
+    r"""Format an alert feature into a readable string."""
+    props = feature["properties"]
+    return f"""
+Event: {props.get('event', 'Unknown')}
+Area: {props.get('areaDesc', 'Unknown')}
+Severity: {props.get('severity', 'Unknown')}
+Description: {props.get('description', 'No description available')}
+Instructions: {props.get('instruction', 'No specific instructions provided')}
+"""
+
+
+@mcp.tool()
+async def get_alerts(state: str) -> str:
+    r"""Get weather alerts for a US state.
+
+    Args:
+        state: Two-letter US state code (e.g. CA, NY)
+    """
+    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
+    data = await make_nws_request(url)
+
+    if not data or "features" not in data:
+        return "Unable to fetch alerts or no alerts found."
+
+    if not data["features"]:
+        return "No active alerts for this state."
+
+    alerts = [format_alert(feature) for feature in data["features"]]
+    return "\n---\n".join(alerts)
+
+
+@mcp.tool()
+async def get_forecast(latitude: float, longitude: float) -> str:
+    r"""Get weather forecast for a location.
+
+    Args:
+        latitude: Latitude of the location
+        longitude: Longitude of the location
+    """
+    # First get the forecast grid endpoint
+    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
+    points_data = await make_nws_request(points_url)
+
+    if not points_data:
+        return "Unable to fetch forecast data for this location."
+
+    # Get the forecast URL from the points response
+    forecast_url = points_data["properties"]["forecast"]
+    forecast_data = await make_nws_request(forecast_url)
+
+    if not forecast_data:
+        return "Unable to fetch detailed forecast."
+
+    # Format the periods into a readable forecast
+    periods = forecast_data["properties"]["periods"]
+    forecasts = []
+    for period in periods[:5]:  # Only show next 5 periods
+        forecast = f"""
+{period['name']}:
+Temperature: {period['temperature']}°{period['temperatureUnit']}
+Wind: {period['windSpeed']} {period['windDirection']}
+Forecast: {period['detailedForecast']}
+"""
+        forecasts.append(forecast)
+
+    return "\n---\n".join(forecasts)
+
+
+def main(transport: str = "stdio"):
+    r"""Weather MCP Server
+
+    This server provides weather-related functionalities implemented via the Model Context Protocol (MCP).
+    It demonstrates how to establish interactions between AI models and external tools using MCP.
+
+    The server supports two modes of operation:
+
+    1. stdio mode (default):
+
+       - Communicates with clients via standard input/output streams, ideal for local command-line usage.
+
+       - Example usage: python mcp_server.py [--transport stdio]
+
+    2. SSE mode (Server-Sent Events):
+
+       - Communicates with clients over HTTP using server-sent events, suitable for persistent network connections.
+
+       - Runs by default at http://127.0.0.1:8000.
+
+       - Example usage: python mcp_server.py --transport sse
+    """  # noqa: E501
+    if transport == 'stdio':
+        mcp.run(transport='stdio')
+    elif transport == 'sse':
+        mcp.run(transport='sse')
+
+
+if __name__ == "__main__":
+    # Hardcoded to use stdio transport mode
+    main("stdio")

requirements.txt

@@ -1,4 +1,4 @@
-camel-ai[all]==0.2.23
+camel-ai[all]==0.2.24
 chunkr-ai>=0.0.41
 docx2markdown>=0.1.1
 gradio>=3.50.2