Make MCP connections concurrency-safe and extend thinking config (#107)

* Fix: MCP Client compete state

* Fix: cot passing

* Upd: test and brand

* Fix: tool refresh

* Fix: deadlock

Author: JohnRichard4096

docs/docs/.vitepress/config.mts

@@ -67,6 +67,9 @@ export default withMermaid({
       themeConfig: {
         // https://vitepress.dev/reference/default-theme-config
         siteTitle: "Amrita Core Docs",
+        search: {
+          provider: "local",
+        },
         nav: [
           { text: "Home", link: "/" },
           { text: "Start", link: "/guide/introduction" },
@@ -420,6 +423,9 @@ export default withMermaid({
       themeConfig: {
         // https://vitepress.dev/reference/default-theme-config
         siteTitle: "Amrita Core 文档",
+        search: {
+          provider: "local",
+        },
         nav: [
           { text: "首页", link: "/zh/" },
           { text: "开始", link: "/zh/guide/introduction" },

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "amrita_core"
-version = "0.10.2"
+version = "0.10.3"
 description = "High performance, flexible, lightweight agent framework."
 readme = "README.md"
 requires-python = ">=3.10,<3.15"

src/amrita_core/adapters/openai.py

@@ -60,8 +60,21 @@ async def call_api(
         completion: ChatCompletion | openai.AsyncStream[ChatCompletionChunk] | None = (
             None
         )
+
+        kwargs.setdefault("extra_body", {})
         if preset.thinking_config is not None:
-            kwargs.update({"reasoning_effort": preset.thinking_config.thinking_effort})
+            if preset.thinking_config.thinking_type is not None:
+                kwargs["extra_body"].setdefault("thinking", {}).setdefault(
+                    "type", preset.thinking_config.thinking_type
+                )
+            if preset.thinking_config.enable_thinking:
+                kwargs["extra_body"].setdefault("enable_thinking", True)
+
+            if preset.thinking_config.thinking_effort:
+                kwargs.update(
+                    {"reasoning_effort": preset.thinking_config.thinking_effort}
+                )
+
         if stream := preset.config.stream:
             kwargs.update({"stream_options": {"include_usage": True}})
         completion = await client.chat.completions.create(
@@ -169,8 +182,19 @@ async def call_tools(
             api_key=key,
             timeout=config.llm.llm_timeout,
         )
+        kwargs.setdefault("extra_body", {})
         if preset.thinking_config is not None:
-            kwargs.update({"reasoning_effort": preset.thinking_config.thinking_effort})
+            if preset.thinking_config.thinking_type is not None:
+                kwargs["extra_body"].setdefault("thinking", {}).setdefault(
+                    "type", preset.thinking_config.thinking_type
+                )
+            if preset.thinking_config.enable_thinking:
+                kwargs["extra_body"].setdefault("enable_thinking", True)
+
+            if preset.thinking_config.thinking_effort:
+                kwargs.update(
+                    {"reasoning_effort": preset.thinking_config.thinking_effort}
+                )
         completion: ChatCompletion = await client.chat.completions.create(
             model=model,
             messages=messages,

src/amrita_core/builtins/agent.py

@@ -886,11 +886,6 @@ def get_category(
     ) -> Literal["workflow"]:
         return "workflow"
 
-
-# TODO: Resolve https://github.com/AmritaBot/AmritaCore/issues/20
-# class CompatibleReActAgentStrategy(BaseReActAgentStrategy):
-
-
 class HybridReActAgentStrategy(BaseReActAgentStrategy):
     """**Hybrid ReAct Agent Strategy optimized for Mixture of Experts (MoE) architecture models.**
 

src/amrita_core/config.py

@@ -1,4 +1,3 @@
-# TODO:分实例的配置
 from __future__ import annotations
 
 import random

src/amrita_core/consts.py

@@ -64,4 +64,8 @@
 ## Mode choice
 
 Calling the `agent_stop` tool indicates "information processing is complete, ready to organize the final response". Immediately after, switch to Daily Conversation Mode and provide the complete response directly, and not to call ANY other tool.
+
+## X-Powered-By
+AmritaCore (Agent framework, see at https://core.amritabot.com)
+AmritaSense (Workflow runtime, see at https://sense.amritabot.com)
 """

src/amrita_core/tools/mcp.py

@@ -39,6 +39,7 @@ class MCPClient:
     server_script: MCP_SERVER_SCRIPT_TYPE
     _close_waitter: asyncio.Future | None = None
     _close_ttl: int
+    _connect_lock: aiologic.Lock
 
     def __init__(
         self,
@@ -59,6 +60,7 @@ def __init__(
         if connection_ttl < -1:
             raise ValueError("connection_ttl must be greater than or equals to -1")
         self._close_ttl = connection_ttl
+        self._connect_lock = aiologic.Lock()
 
     async def __aenter__(self) -> Self:
         self._close_waitter = None
@@ -105,25 +107,36 @@ async def simple_call(self, tool_name: str, data: dict[str, Any]) -> str:
             await self._close()
 
     async def _connect(self, update_tools: bool = False):
-        """Connect to MCP Server
+        """Connect to MCP Server (idempotent, safe for concurrent calls)
         Args:
             update_tools (bool, optional): whether to update the tool list. Defaults to False.
+                When True, tools are always refreshed from the server, even if already connected.
         """
+        # Cancel any pending close-waiter outside the lock to avoid deadlock:
+        # _clean_waitter cancels the waiter task whose _close() also needs _connect_lock.
         await self._clean_waitter()
-        if self.mcp_client is not None:
-            raise RuntimeError("MCP Server is already connected!")
-
-        server_script: MCP_SERVER_SCRIPT_TYPE = self.server_script
-        self.mcp_client = Client(server_script)
-        await self.mcp_client.__aenter__()
-        logger.info(f"Successfully connected to MCP Server@{server_script}")
-        if update_tools or not self.tools:
-            self.tools = [
-                MCPToolSchema.model_validate(i.model_dump())
-                for i in await self.mcp_client.list_tools()
-            ]
-            logger.info(f"Available tools: {[tool.name for tool in self.tools]}")
-            self._cast_tool_to_amrita()
+        async with self._connect_lock:
+            if self.mcp_client is not None:
+                if update_tools:
+                    await self._refresh_tools()
+                return
+
+            server_script: MCP_SERVER_SCRIPT_TYPE = self.server_script
+            self.mcp_client = Client(server_script)
+            await self.mcp_client.__aenter__()
+            logger.info(f"Successfully connected to MCP Server@{server_script}")
+            if update_tools or not self.tools:
+                await self._refresh_tools()
+
+    async def _refresh_tools(self) -> None:
+        """Refresh the cached tool list from the connected MCP server."""
+        assert self.mcp_client is not None, "Cannot refresh tools: not connected"
+        self.tools = [
+            MCPToolSchema.model_validate(i.model_dump())
+            for i in await self.mcp_client.list_tools()
+        ]
+        logger.info(f"Available tools: {[tool.name for tool in self.tools]}")
+        self._cast_tool_to_amrita()
 
     async def _clean_waitter(self):
         if self._close_waitter is not None:
@@ -158,9 +171,10 @@ async def close_no_wait(self):
 
     async def _close(self) -> None:
         """Close connection"""
-        if self.mcp_client:
-            await self.mcp_client.__aexit__(None, None, None)
-            self.mcp_client = None
+        async with self._connect_lock:
+            if self.mcp_client:
+                await self.mcp_client.__aexit__(None, None, None)
+                self.mcp_client = None
 
     def _format_tools_for_openai(self):
         """Convert MCP tool format to OpenAI tool format"""

src/amrita_core/types/preset.py

@@ -32,7 +32,9 @@ class ModelConfig(BaseModel):
     )
 
 
-class ThinkingConfig(BaseModel):
+class ThinkingConfig(
+    BaseModel
+):  # TODO: use model-adapter specific config like the structure of `Content`.
     """Thinking/reasoning configuration for a model preset."""
 
     thinking_type: Literal["enabled", "disabled"] | None = Field(

tests/test_mcp.py

@@ -88,12 +88,11 @@ async def test_simple_call_with_exception(self, mcp_client):
             assert "Test error" in error_data["error"]
 
     @pytest.mark.asyncio
-    async def test_connect_already_connected(self, mcp_client):
+    async def test_connect_already_connected_suppress(self, mcp_client):
         """Test _connect when already connected"""
         mcp_client.mcp_client = AsyncMock()
 
-        with pytest.raises(RuntimeError, match="MCP Server is already connected!"):
-            await mcp_client._connect()
+        await mcp_client._connect()
 
 
 class TestMultiClientManager: