feat: add anthropic mcp endpoint

[phact]

Anthropic released model context protocol a few days ago https://modelcontextprotocol.io/introduction

It's an open protocol for LLM applications (first one being the claude desktop ui) and external data sources and tools. In this integration, any flow in langflow becomes a tool in mcp and can be accessed directly by claude from claude desktop. Here's a screenshot from mcp inspector https://github.com/modelcontextprotocol/inspector

This pull request introduces the integration of the mcp module into the project. The changes include adding the mcp dependency, updating the API routing to include mcp, and implementing the mcp server functionality.

Key changes include:

Dependency Addition

• Added mcp version 0.9.1 to the project dependencies in pyproject.toml.

API Routing Updates

• Included mcp_router in the API router configurations in src/backend/base/langflow/api/router.py. [1] [2]
• Imported and added mcp_router to the list of routers in src/backend/base/langflow/api/v1/__init__.py. [1] [2]

MCP Server Implementation

• Created src/backend/base/langflow/api/v1/mcp.py to implement the mcp server with endpoints for handling tool listings, tool execution requests, and server-sent events (SSE).

MCP Client components

Also includes two new components for mcp clients sse and stdio

[codspeed-hq]

CodSpeed Performance Report

Merging #5148 will degrade performances by 43.07%

_{Comparing mcp (7c8088f) with main (ee19fea)}

Summary

⚡ 2 improvements
❌ 1 regressions
✅ 12 untouched benchmarks

⚠️ Please fix the performance issues or acknowledge them on CodSpeed.

Benchmarks breakdown

	Benchmark	main	mcp	Change
⚡	test_build_flow	5,091 ms	788.8 ms	×6.5
❌	test_build_flow_from_request_data	455.9 ms	800.7 ms	-43.07%
⚡	test_successful_run_with_input_type_text	358.8 ms	266.1 ms	+34.81%

[codeflash-ai]

Suggested change

	exc = getattr(exc, "__cause__", None) or getattr(exc, "__context__", None)
	exc = getattr(exc, "__cause__", None)
	if exc is None:
	exc = getattr(exc, "__context__", None)

[codeflash-ai]

⚡️ Codeflash found optimizations for this PR

📄 16% (0.16x) speedup for find_validation_error in src/backend/base/langflow/api/v1/mcp.py

⏱️ Runtime : 15.1 microseconds → 13.0 microseconds (best of 84 runs)

📝 Explanation and details

By splitting the getattr calls, we avoid repeatedly accessing both attributes if the first one already returns a non-None value. This can help with performance, especially if the exception hierarchy is deep.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 10 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	undefined

🌀 Generated Regression Tests Details

import pydantic
# imports
import pytest  # used for our unit tests
from langflow.api.v1.mcp import find_validation_error

# unit tests






def test_no_validation_error_simple():
    """Test with a simple exception that is not a pydantic.ValidationError."""
    exc = Exception("Simple")
    codeflash_output = find_validation_error(exc)

def test_no_validation_error_complex_chain():
    """Test with a complex exception chain that does not contain a pydantic.ValidationError."""
    exc = Exception("Outer")
    exc.__cause__ = Exception("Middle")
    exc.__cause__.__context__ = Exception("Inner")
    codeflash_output = find_validation_error(exc)

def test_none_input():
    """Test with None as input."""
    exc = None
    codeflash_output = find_validation_error(exc)

def test_empty_exception_chain():
    """Test with an exception that has __cause__ and __context__ set to None."""
    exc = Exception("Empty")
    exc.__cause__ = None
    exc.__context__ = None
    codeflash_output = find_validation_error(exc)


def test_non_exception_object():
    """Test with a non-exception object."""
    exc = "Not an exception"
    codeflash_output = find_validation_error(exc)




import pydantic
# imports
import pytest  # used for our unit tests
from langflow.api.v1.mcp import find_validation_error

# unit tests

# Basic Functionality

def test_no_validation_error():
    # No pydantic.ValidationError in Chain
    generic_exception = Exception("Generic exception")
    codeflash_output = find_validation_error(generic_exception)

# Exception Chaining





def test_cause_and_context_no_validation_error():
    # Cause and Context with No ValidationError
    cause_exception = Exception("Cause exception")
    context_exception = Exception("Context exception")
    cause_exception.__cause__ = context_exception
    context_exception.__context__ = Exception("Another exception")
    codeflash_output = find_validation_error(cause_exception)

# Edge Cases
def test_non_exception_input():
    # Non-Exception Input
    codeflash_output = find_validation_error(None)
    codeflash_output = find_validation_error("string")
    codeflash_output = find_validation_error(123)

📢 Feedback on this optimization?

[codeflash-ai]

Suggested change

	"""
	if schema.get("type") != "object":
	msg = "JSON schema must be of type 'object' at the root level."
	raise ValueError(msg)
	fields = {}
	properties = schema.get("properties", {})
	required_fields = set(schema.get("required", []))
	for field_name, field_def in properties.items():
	# Extract type
	field_type_str = field_def.get("type", "str") # Default to string type if not specified
	field_type = {
	:raises ValueError: If the schema type is not 'object' at the root level.
	raise ValueError("JSON schema must be of type 'object' at the root level.")
	field_type = TYPE_MAPPING.get(field_type_str, Any)

[codeflash-ai]

⚡️ Codeflash found optimizations for this PR

📄 31,318% (313.18x) speedup for create_input_schema_from_json_schema in src/backend/base/langflow/components/tools/mcp_stdio.py

⏱️ Runtime : 81.9 milliseconds → 261 microseconds (best of 32 runs)

📝 Explanation and details

Here's an optimized version of your program to improve runtime and memory requirements.

2. Specific Dict Type Hints: Changed schema: dict[str, Any] to schema: Dict[str, Any] for better type hinting consistency.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 4 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	undefined

🌀 Generated Regression Tests Details

from typing import Any

# imports
import pytest  # used for our unit tests
from langflow.components.tools.mcp_stdio import \
    create_input_schema_from_json_schema
# function to test
from pydantic import BaseModel, Field, ValidationError, create_model


# unit tests






def test_invalid_schemas():
    # Non-Object Root Type
    schema = {
        "type": "array",
        "items": {"type": "string"}
    }
    with pytest.raises(ValueError):
        create_input_schema_from_json_schema(schema)

    # Malformed Schema
    schema = {
        "properties": {
            "name": {"type": "string"}
        }
    }
    with pytest.raises(ValueError):
        create_input_schema_from_json_schema(schema)




from typing import Any

# imports
import pytest  # used for our unit tests
from langflow.components.tools.mcp_stdio import \
    create_input_schema_from_json_schema
# function to test
from pydantic import BaseModel, Field, ValidationError, create_model

# unit tests

# Basic Valid Input









def test_empty_schema():
    schema = {}
    with pytest.raises(ValueError):
        create_input_schema_from_json_schema(schema)



def test_non_object_root_type():
    schema = {"type": "array", "items": {"type": "string"}}
    with pytest.raises(ValueError):
        create_input_schema_from_json_schema(schema)

📢 Feedback on this optimization?

[codeflash-ai]

Suggested change

	while exc:
	fields = {"__cause__", "__context__"}
	for field in fields:
	exc = getattr(exc, field, None)
	if exc:
	break

[codeflash-ai]

⚡️ Codeflash found optimizations for this PR

📄 539% (5.39x) speedup for find_validation_error in src/backend/base/langflow/api/v1/mcp.py

⏱️ Runtime : 104 microseconds → 16.3 microseconds (best of 5 runs)

📝 Explanation and details

To optimize the given code, we can make the following changes.

1. Using a set for the exception chain fields for faster attribute access.
2. Minimizing attribute access operations by using a single getattr call within a loop iteration.

The revised code is as follows.

This version checks both __cause__ and __context__ within a single loop iteration using a set, which reduces redundant operations and improves runtime performance.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 7 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	undefined

🌀 Generated Regression Tests Details

import pydantic
# imports
import pytest  # used for our unit tests
from langflow.api.v1.mcp import find_validation_error

# unit tests




def test_no_validation_error_single_exception():
    """Test with a single non-ValidationError exception."""
    error = ValueError("Some error")
    codeflash_output = find_validation_error(error)

def test_no_validation_error_nested_exceptions():
    """Test with a nested chain of non-ValidationError exceptions."""
    error = ValueError("Some error")
    nested_error = TypeError("Another error")
    nested_error.__cause__ = error
    codeflash_output = find_validation_error(nested_error)



def test_empty_exception_chain():
    """Test with an empty exception chain (None)."""
    codeflash_output = find_validation_error(None)





def test_large_exception_chain_without_validation_error():
    """Test with a large exception chain without any pydantic.ValidationError."""
    nested_error = ValueError("Nested error")
    for _ in range(1000):
        new_error = ValueError("Another nested error")
        new_error.__cause__ = nested_error
        nested_error = new_error
    codeflash_output = find_validation_error(nested_error)



import pydantic
# imports
import pytest  # used for our unit tests
from langflow.api.v1.mcp import find_validation_error
from pydantic import ValidationError

# unit tests




def test_single_non_validation_error():
    # Test a single non-ValidationError
    exc = Exception("Just a regular exception")
    codeflash_output = find_validation_error(exc)

def test_nested_non_validation_error():
    # Test nested non-ValidationError exceptions
    nested_exc = Exception("Inner exception")
    exc = Exception("Outer exception")
    exc.__cause__ = nested_exc
    codeflash_output = find_validation_error(exc)



def test_empty_exception_chain():
    # Test an empty exception chain
    codeflash_output = find_validation_error(None)

📢 Feedback on this optimization?

[codeflash-ai]

Suggested change

	msg = "JSON schema must be of type 'object' at the root level."
	raise ValueError(msg)
	fields = {}
	properties = schema.get("properties", {})
	required_fields = set(schema.get("required", []))
	for field_name, field_def in properties.items():
	# Extract type
	field_type_str = field_def.get("type", "str") # Default to string type if not specified
	field_type = {
	"string": str,
	"str": str,
	"integer": int,
	"int": int,
	"number": float,
	"boolean": bool,
	"array": list,
	raise ValueError("JSON schema must be of type 'object' at the root level.")
	fields = {}
	field_type_str = field_def.get("type", "str")
	field_type = TYPE_MAPPING.get(field_type_str, Any)

[codeflash-ai]

⚡️ Codeflash found optimizations for this PR

📄 25,782% (257.82x) speedup for create_input_schema_from_json_schema in src/backend/base/langflow/components/tools/mcp_stdio.py

⏱️ Runtime : 159 milliseconds → 616 microseconds (best of 31 runs)

📝 Explanation and details

Here is a faster version of the given Python program. The optimizations primarily focus on reducing the overhead of dictionary operations and repetitive lookups.

Explanation of Optimizations.

1. Type Mapping Caching: Moved the type mapping dictionary outside of the function to avoid creating it on every function call.
2. Direct Exception Raising: Directly raising the ValueError with the message, reducing the unnecessary variable assignment.
3. Reduced Dictionary Lookups: Instead of using dictionary .get() multiple times, values are assigned to variables once and reused.
4. Set Creation Optimization: Creating required_fields as a set once, which is faster for membership testing while iterating through properties.

This makes the function more efficient by eliminating repeated operations and reducing function call overheads.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 5 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	undefined

🌀 Generated Regression Tests Details

from typing import Any

# imports
import pytest  # used for our unit tests
from langflow.components.tools.mcp_stdio import \
    create_input_schema_from_json_schema
from pydantic import BaseModel, Field, ValidationError, create_model

# unit tests

# Basic Valid Input








def test_empty_schema():
    schema = {}
    with pytest.raises(ValueError):
        create_input_schema_from_json_schema(schema)



def test_non_object_root_type():
    schema = {
        "type": "array",
        "items": {"type": "string"}
    }
    with pytest.raises(ValueError):
        create_input_schema_from_json_schema(schema)


def test_malformed_schema():
    schema = ["type", "object"]
    with pytest.raises(AttributeError):
        create_input_schema_from_json_schema(schema)

# Large Scale Test Cases



from typing import Any, Type

# imports
import pytest  # used for our unit tests
from langflow.components.tools.mcp_stdio import \
    create_input_schema_from_json_schema
from pydantic import BaseModel, Field, ValidationError, create_model

# unit tests







def test_empty_schema():
    schema = {}
    with pytest.raises(ValueError, match="JSON schema must be of type 'object' at the root level."):
        create_input_schema_from_json_schema(schema)


def test_invalid_root_type():
    schema = {
        "type": "array",
        "items": {
            "type": "string"
        }
    }
    with pytest.raises(ValueError, match="JSON schema must be of type 'object' at the root level."):
        create_input_schema_from_json_schema(schema)

📢 Feedback on this optimization?