feat(mcp adapter): add adapter to use mcp tools

Author: NicholasGoh

backend/api/core/dependencies.py

@@ -1,7 +1,10 @@
-from typing import Annotated
+from contextlib import asynccontextmanager
+from typing import Annotated, AsyncGenerator
 
 from fastapi import Depends
 from langchain_openai import ChatOpenAI
+from mcp import ClientSession
+from mcp.client.sse import sse_client
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 
 from api.core.config import settings
@@ -28,3 +31,14 @@ def get_engine() -> AsyncEngine:
 
 
 EngineDep = Annotated[AsyncEngine, Depends(get_engine)]
+
+
+@asynccontextmanager
+async def mcp_sse_client() -> AsyncGenerator[ClientSession, None]:
+    async with sse_client(f"http://mcp:{settings.mcp_server_port}/sse") as (
+        read_stream,
+        write_stream,
+    ):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+            yield session

backend/api/pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "fastapi[standard]==0.115.11",
     "langchain==0.3.6",
     "langchain-community==0.3.4",
+    "langchain-mcp-adapters>=0.0.9",
     "langchain-openai==0.2.3",
     "langfuse==2.60.2",
     "langgraph==0.2.39",

backend/api/routers/mcps.py

@@ -1,27 +1,14 @@
-from contextlib import asynccontextmanager
 from typing import Iterable
 
 from fastapi import APIRouter
-from mcp import ClientSession, types
-from mcp.client.sse import sse_client
+from mcp import types
 
-from api.core.config import settings
+from api.core.dependencies import mcp_sse_client
 from shared_mcp.models import ToolRequest
 
 router = APIRouter(prefix="/mcps", tags=["mcps"])
 
 
-@asynccontextmanager
-async def mcp_sse_client():
-    async with sse_client(f"http://mcp:{settings.mcp_server_port}/sse") as (
-        read_stream,
-        write_stream,
-    ):
-        async with ClientSession(read_stream, write_stream) as session:
-            await session.initialize()
-            yield session
-
-
 @router.get("/list-tools")
 async def list_tools() -> Iterable[types.Tool]:
     """