# Directory Structure
```
├── .github
│ └── workflows
│ ├── canary.yml
│ ├── publish.yml
│ └── test.yml
├── .gitignore
├── .python-version
├── .vscode
│ └── settings.json
├── LICENSE
├── pyproject.toml
├── README.md
├── src
│ └── langchain_mcp
│ ├── __init__.py
│ ├── py.typed
│ └── toolkit.py
├── tests
│ ├── conftest.py
│ ├── demo.py
│ ├── integration_tests
│ │ ├── __init__.py
│ │ └── test_tool.py
│ └── unit_tests
│ ├── __init__.py
│ └── test_tool.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.10
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# langchain-mcp
*Note: langchain now has a more official implementation [langchain-mcp-adapters](https://github.com/langchain-ai/langchain-mcp-adapters)*
[Model Context Protocol](https://modelcontextprotocol.io) tool calling support in LangChain.
Create a `langchain_mcp.MCPToolkit` with an `mcp.ClientSession`,
then `await toolkit.initialize()` and `toolkit.get_tools()` to get the list of `langchain_core.tools.BaseTool`s.
Example:
https://github.com/rectalogic/langchain-mcp/blob/8fa8445a24755bf91789f52718c32361ed916f46/tests/demo.py#L34-L43
## Demo
You can run the demo against [Groq](https://groq.com/) `llama-3.1-8b-instant`:
```sh-session
$ export GROQ_API_KEY=xxx
$ uv run tests/demo.py "Read and summarize the file ./LICENSE"
Secure MCP Filesystem Server running on stdio
Allowed directories: [ '/users/aw/projects/rectalogic/langchain-mcp' ]
The file ./LICENSE is a MIT License agreement. It states that the software is provided "as is" without warranty and that the authors and copyright holders are not liable for any claims, damages, or other liability arising from the software or its use.
```
```
--------------------------------------------------------------------------------
/tests/integration_tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/tests/unit_tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/langchain_mcp/__init__.py:
--------------------------------------------------------------------------------
```python
# Copyright (C) 2024 Andrew Wason
# SPDX-License-Identifier: MIT
from .toolkit import MCPToolkit # noqa: F401
```
--------------------------------------------------------------------------------
/.vscode/settings.json:
--------------------------------------------------------------------------------
```json
{
"[python]": {
"editor.formatOnSave": true,
"editor.codeActionsOnSave": {
"source.organizeImports": "explicit"
},
"editor.defaultFormatter": "charliermarsh.ruff"
}
}
```
--------------------------------------------------------------------------------
/tests/unit_tests/test_tool.py:
--------------------------------------------------------------------------------
```python
# Copyright (C) 2024 Andrew Wason
# SPDX-License-Identifier: MIT
import pytest
from langchain_tests.unit_tests import ToolsUnitTests
@pytest.mark.usefixtures("mcptool")
class TestMCPToolUnit(ToolsUnitTests):
@property
def tool_constructor(self):
return self.tool
@property
def tool_invoke_params_example(self) -> dict:
return {"path": "LICENSE"}
```
--------------------------------------------------------------------------------
/tests/integration_tests/test_tool.py:
--------------------------------------------------------------------------------
```python
# Copyright (C) 2024 Andrew Wason
# SPDX-License-Identifier: MIT
import pytest
from langchain_tests.integration_tests import ToolsIntegrationTests
@pytest.mark.usefixtures("mcptool")
class TestMCPToolIntegration(ToolsIntegrationTests):
@property
def tool_constructor(self):
return self.tool
@property
def tool_invoke_params_example(self) -> dict:
return {"path": "LICENSE"}
```
--------------------------------------------------------------------------------
/.github/workflows/canary.yml:
--------------------------------------------------------------------------------
```yaml
name: Canary tests
on:
workflow_call:
schedule:
- cron: "* * * * 0"
permissions:
contents: read
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.10", "3.11", "3.12", "3.13"]
steps:
- uses: actions/checkout@v4
- name: Install uv
uses: astral-sh/setup-uv@v2
with:
enable-cache: false
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Run tests with upgraded packages
run: uv run --upgrade --python ${{ matrix.python-version }} --python-preference only-system pytest
- name: Mypy
run: |
uv run --upgrade --python ${{ matrix.python-version }} --python-preference only-system mypy
```
--------------------------------------------------------------------------------
/.github/workflows/test.yml:
--------------------------------------------------------------------------------
```yaml
name: Test
on: [push, pull_request, workflow_call]
permissions:
contents: read
jobs:
test:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11", "3.12", "3.13"]
steps:
- uses: actions/checkout@v4
- name: Install uv
uses: astral-sh/setup-uv@v2
with:
enable-cache: true
cache-dependency-glob: "uv.lock"
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Run tests
run: uv run --isolated --locked --python ${{ matrix.python-version }} --python-preference only-system pytest
- name: Lint
run: |
uv run --python ${{ matrix.python-version }} --python-preference only-system ruff format --check
uv run --python ${{ matrix.python-version }} --python-preference only-system ruff check
- name: Mypy
run: |
uv run --python ${{ matrix.python-version }} --python-preference only-system mypy
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "langchain-mcp"
version = "0.2.1"
description = "Model Context Protocol tool calling support for LangChain"
readme = "README.md"
authors = [{ name = "Andrew Wason", email = "[email protected]" }]
requires-python = ">=3.10"
dependencies = [
"langchain-core~=0.3.37",
"mcp~=1.0",
"pydantic~=2.10",
"typing-extensions~=4.12",
]
classifiers = ["License :: OSI Approved :: MIT License"]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[dependency-groups]
dev = [
"langchain-tests~=0.3",
"pytest~=8.3",
"pytest-asyncio~=0.24",
"pytest-socket~=0.7",
"ruff~=0.8",
"mypy~=1.13",
"langchain-groq~=0.2",
]
[project.urls]
Repository = "https://github.com/rectalogic/langchain-mcp"
Issues = "https://github.com/rectalogic/langchain-mcp/issues"
Related = "https://modelcontextprotocol.io/"
[tool.ruff]
target-version = "py310"
line-length = 120
[tool.ruff.format]
docstring-code-format = true
[tool.ruff.lint]
select = [
# flake8-2020
"YTT",
# flake8-bandit
"S",
# flake8-bugbear
"B",
# flake8-builtins
"A",
# Pyflakes
"F",
# Pycodestyle
"E",
"W",
# isort
"I",
# flake8-no-pep420
"INP",
# pyupgrade
"UP",
]
[tool.ruff.lint.per-file-ignores]
"tests/*" = ["S", "INP001"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "class"
[tool.mypy]
disallow_untyped_defs = true
warn_unused_configs = true
warn_redundant_casts = true
warn_unused_ignores = true
strict_equality = true
no_implicit_optional = true
show_error_codes = true
files = ["src/**/*.py", "tests/demo.py"]
plugins = "pydantic.mypy"
```
--------------------------------------------------------------------------------
/tests/demo.py:
--------------------------------------------------------------------------------
```python
# Copyright (C) 2024 Andrew Wason
# SPDX-License-Identifier: MIT
import asyncio
import pathlib
import sys
import typing as t
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import BaseTool
from langchain_groq import ChatGroq
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp import MCPToolkit
async def run(tools: list[BaseTool], prompt: str) -> str:
model = ChatGroq(model_name="llama-3.1-8b-instant", stop_sequences=None) # requires GROQ_API_KEY
tools_map = {tool.name: tool for tool in tools}
tools_model = model.bind_tools(tools)
messages: list[BaseMessage] = [HumanMessage(prompt)]
ai_message = t.cast(AIMessage, await tools_model.ainvoke(messages))
messages.append(ai_message)
for tool_call in ai_message.tool_calls:
selected_tool = tools_map[tool_call["name"].lower()]
tool_msg = await selected_tool.ainvoke(tool_call)
messages.append(tool_msg)
return await (tools_model | StrOutputParser()).ainvoke(messages)
async def main(prompt: str) -> None:
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)
if __name__ == "__main__":
prompt = sys.argv[1] if len(sys.argv) > 1 else "Read and summarize the file ./LICENSE"
asyncio.run(main(prompt))
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
# Copyright (C) 2024 Andrew Wason
# SPDX-License-Identifier: MIT
from unittest import mock
import pytest
from langchain_tests.integration_tests import ToolsIntegrationTests
from mcp import ClientSession, ListToolsResult, Tool
from mcp.types import CallToolResult, TextContent
from langchain_mcp import MCPToolkit
@pytest.fixture(scope="class")
def mcptoolkit(request):
session_mock = mock.AsyncMock(spec=ClientSession)
session_mock.list_tools.return_value = ListToolsResult(
tools=[
Tool(
name="read_file",
description=(
"Read the complete contents of a file from the file system. Handles various text encodings "
"and provides detailed error messages if the file cannot be read. "
"Use this tool when you need to examine the contents of a single file. "
"Only works within allowed directories."
),
inputSchema={
"type": "object",
"properties": {"path": {"type": "string"}},
"required": ["path"],
"additionalProperties": False,
"$schema": "http://json-schema.org/draft-07/schema#",
},
)
]
)
session_mock.call_tool.return_value = CallToolResult(
content=[TextContent(type="text", text="MIT License\n\nCopyright (c) 2024 Andrew Wason\n")],
isError=False,
)
toolkit = MCPToolkit(session=session_mock)
yield toolkit
if issubclass(request.cls, ToolsIntegrationTests):
session_mock.call_tool.assert_called_with("read_file", arguments={"path": "LICENSE"})
@pytest.fixture(scope="class")
async def mcptool(request, mcptoolkit):
await mcptoolkit.initialize()
tool = mcptoolkit.get_tools()[0]
request.cls.tool = tool
yield tool
```
--------------------------------------------------------------------------------
/.github/workflows/publish.yml:
--------------------------------------------------------------------------------
```yaml
name: Publish Python Package and GitHub Release
on:
push:
tags:
- '**'
jobs:
call-test:
uses: ./.github/workflows/test.yml
build:
name: Build distribution package
runs-on: ubuntu-latest
needs: [call-test]
steps:
- uses: actions/checkout@v4
- name: Install uv
uses: astral-sh/setup-uv@v2
with:
enable-cache: true
cache-dependency-glob: "uv.lock"
- name: Set up Python 3.12
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install dependencies
run: |
uv sync --locked --dev --python 3.12 --python-preference only-system
version=$(uv run --locked python -c 'from importlib.metadata import version; print(version("langchain_mcp"))')
if [ "$version" != "${{ github.ref_name }}" ]; then
echo "Built version $version does not match tag ${{ github.ref_name }}, aborting"
exit 1
fi
- name: Build
run: |
uv build
- name: Store the distribution packages
uses: actions/upload-artifact@v4
with:
name: python-package-distribution
path: dist/
publish:
name: Publish Python distribution package to PyPI
runs-on: ubuntu-latest
needs:
- build
environment: release
permissions:
id-token: write
steps:
- name: Download dist
uses: actions/download-artifact@v4
with:
name: python-package-distribution
path: dist/
- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
github-release:
name: Sign the Python distribution package with Sigstore and upload to GitHub Release
needs:
- publish
runs-on: ubuntu-latest
permissions:
contents: write
id-token: write
steps:
- name: Download dists
uses: actions/download-artifact@v4
with:
name: python-package-distribution
path: dist/
- name: Sign with Sigstore
uses: sigstore/[email protected]
with:
inputs: >-
./dist/*.tar.gz
./dist/*.whl
- name: Create GitHub Release
env:
GITHUB_TOKEN: ${{ github.token }}
run: >-
gh release create
'${{ github.ref_name }}'
--repo '${{ github.repository }}'
--notes ""
- name: Upload artifact signatures to GitHub Release
env:
GITHUB_TOKEN: ${{ github.token }}
run: >-
gh release upload
'${{ github.ref_name }}' dist/**
--repo '${{ github.repository }}'
```
--------------------------------------------------------------------------------
/src/langchain_mcp/toolkit.py:
--------------------------------------------------------------------------------
```python
# Copyright (C) 2024 Andrew Wason
# SPDX-License-Identifier: MIT
import asyncio
import warnings
from collections.abc import Callable
import pydantic
import pydantic_core
import typing_extensions as t
from langchain_core.tools.base import BaseTool, BaseToolkit, ToolException
from mcp import ClientSession, ListToolsResult
from mcp.types import EmbeddedResource, ImageContent, TextContent
class MCPToolkit(BaseToolkit):
"""
MCP server toolkit
"""
session: ClientSession
"""The MCP session used to obtain the tools"""
_tools: ListToolsResult | None = None
model_config = pydantic.ConfigDict(arbitrary_types_allowed=True)
async def initialize(self) -> None:
"""Initialize the session and retrieve tools list"""
if self._tools is None:
await self.session.initialize()
self._tools = await self.session.list_tools()
@t.override
def get_tools(self) -> list[BaseTool]:
if self._tools is None:
raise RuntimeError("Must initialize the toolkit first")
return [
MCPTool(
session=self.session,
name=tool.name,
description=tool.description or "",
args_schema=tool.inputSchema,
)
# list_tools returns a PaginatedResult, but I don't see a way to pass the cursor to retrieve more tools
for tool in self._tools.tools
]
class MCPTool(BaseTool):
"""
MCP server tool
"""
session: ClientSession
handle_tool_error: bool | str | Callable[[ToolException], str] | None = True
response_format: t.Literal["content", "content_and_artifact"] = "content_and_artifact"
@t.override
def _run(self, *args: t.Any, **kwargs: t.Any) -> tuple[str, list[ImageContent | EmbeddedResource]]:
warnings.warn(
"Invoke this tool asynchronousely using `ainvoke`. This method exists only to satisfy standard tests.",
stacklevel=1,
)
return asyncio.run(self._arun(*args, **kwargs))
@t.override
async def _arun(self, *args: t.Any, **kwargs: t.Any) -> tuple[str, list[ImageContent | EmbeddedResource]]:
result = await self.session.call_tool(self.name, arguments=kwargs)
if result.isError:
raise ToolException(pydantic_core.to_json(result.content).decode())
text_content = [block for block in result.content if isinstance(block, TextContent)]
artifacts = [block for block in result.content if not isinstance(block, TextContent)]
return pydantic_core.to_json(text_content).decode(), artifacts
```