Langgraph MCP Tool

LangGraph MCP Tool Skill

Purpose

This skill helps create LangChain-compatible tools from Databricks MCP (Model Context Protocol) servers. These tools can be used with any AI agent framework (LangGraph, LangChain, custom agents). The focus is on tool creation, not agent implementation.

When to Activate

This skill should activate when the user requests:

• "Create MCP tools for my agent"
• "Connect to Databricks managed MCP"
• "Get tools from Vector Search MCP"
• "Use Genie Space as tools"
• "Connect to external MCP server as LangChain tools"
• "Set up MCP tools from Unity Catalog functions"

Prerequisites

• Python 3.12+
• Databricks CLI authenticated (databricks auth login)
• Serverless compute enabled in workspace
• Unity Catalog enabled

Installation

pip install "databricks-mcp>=0.6.0" "databricks-langchain>=0.13.0" "mcp>=1.9" langchain-mcp-adapters

Core Packages

Key Classes

Available Managed MCP Servers

Pattern 1: Single Managed MCP Server

Connect to one Databricks managed MCP server and get LangChain tools.

from databricks_mcp import DatabricksMCPClient

async def get_vector_search_tools():
    """Get tools from Vector Search MCP server."""
    async with DatabricksMCPClient(
        server_url="/api/2.0/mcp/vector-search/my_catalog/my_schema"
    ) as client:
        # Get all tools as LangChain-compatible tools
        tools = await client.get_tools()
        return tools

# Usage
import asyncio
tools = asyncio.run(get_vector_search_tools())
# tools can now be used with any LangChain-compatible agent

Pattern 2: Multi-Server MCP Client

Connect to multiple Databricks MCP servers and aggregate tools.

from databricks_mcp import DatabricksMultiServerMCPClient, DatabricksMCPServer

async def get_multi_server_tools():
    """Get tools from multiple MCP servers."""
    servers = [
        DatabricksMCPServer(
            name="vector_search",
            server_url="/api/2.0/mcp/vector-search/catalog/schema"
        ),
        DatabricksMCPServer(
            name="genie",
            server_url="/api/2.0/mcp/genie/my_genie_space_id"
        ),
        DatabricksMCPServer(
            name="uc_functions",
            server_url="/api/2.0/mcp/functions/catalog/schema"
        ),
    ]

    async with DatabricksMultiServerMCPClient(servers=servers) as client:
        # Get all tools from all servers
        tools = await client.get_tools()
        return tools

# Usage
import asyncio
tools = asyncio.run(get_multi_server_tools())

Pattern 3: Vector Search with Meta Parameters

Fine-tune Vector Search queries using _meta parameters.

from databricks_mcp import DatabricksMCPClient

async def get_vector_search_tools_with_meta():
    """Get Vector Search tools with meta parameter configuration."""
    async with DatabricksMCPClient(
        server_url="/api/2.0/mcp/vector-search/my_catalog/my_schema"
    ) as client:
        tools = await client.get_tools()
        return tools

# When calling the tool, include _meta parameters:
# tool_input = {
#     "query": "machine learning best practices",
#     "_meta": {
#         "num_results": 10,                    # Number of results (default: 5)
#         "filters": {"category": "ML"},        # Filter conditions
#         "query_type": "hybrid",               # "ann" | "hybrid"
#         "score_threshold": 0.7,               # Minimum similarity score
#         "columns": ["title", "content"],      # Columns to return
#     }
# }

Available Meta Parameters for Vector Search

Pattern 4: External MCP Servers via Proxy

Connect to external MCP servers through Databricks proxy.

from databricks_mcp import DatabricksMCPClient

async def get_external_mcp_tools():
    """Get tools from external MCP server via Databricks proxy."""
    # External servers are accessed via Unity Catalog connections
    async with DatabricksMCPClient(
        server_url="/api/2.0/mcp/external/my_github_connection"
    ) as client:
        tools = await client.get_tools()
        return tools

import asyncio
tools = asyncio.run(get_external_mcp_tools())

Pattern 5: External MCP Servers (Direct)

Connect directly to external MCP servers using langchain-mcp-adapters.

import os
from langchain_mcp_adapters.client import MultiServerMCPClient

async def get_external_tools_direct():
    """Connect directly to external MCP servers."""
    async with MultiServerMCPClient(
        {
            "github": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-github"],
                "env": {"GITHUB_TOKEN": os.getenv("GITHUB_TOKEN")},
            },
            "filesystem": {
                "command": "npx",
                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/allowed"],
            },
        }
    ) as client:
        tools = client.get_tools()
        return tools

import asyncio
tools = asyncio.run(get_external_tools_direct())

Pattern 6: OAuth Authentication

Use OAuth for authentication instead of PAT tokens.

import os
from databricks_mcp import DatabricksMCPClient
from databricks_mcp.auth import OAuthProvider

async def get_tools_with_oauth():
    """Get MCP tools using OAuth authentication."""
    # OAuth provider handles token refresh automatically
    auth_provider = OAuthProvider(
        client_id=os.getenv("DATABRICKS_CLIENT_ID"),
        client_secret=os.getenv("DATABRICKS_CLIENT_SECRET"),
        # Or use service principal authentication
    )

    async with DatabricksMCPClient(
        server_url="/api/2.0/mcp/vector-search/catalog/schema",
        auth_provider=auth_provider
    ) as client:
        tools = await client.get_tools()
        return tools

Pattern 7: Get Resources for Deployment

Get Databricks resources required for model serving deployment.

from databricks_mcp import DatabricksMultiServerMCPClient, DatabricksMCPServer

async def get_deployment_resources():
    """Get Databricks resources for deployment configuration."""
    servers = [
        DatabricksMCPServer(
            name="vector_search",
            server_url="/api/2.0/mcp/vector-search/catalog/schema"
        ),
    ]

    async with DatabricksMultiServerMCPClient(servers=servers) as client:
        # Get resources needed for Model Serving deployment
        resources = client.get_databricks_resources()
        # Returns: {"vector_search_indexes": [...], "genie_spaces": [...], etc.}
        return resources

Using Resources in Model Serving

import mlflow
from mlflow.models.resources import DatabricksVectorSearchIndex

# Get resources from MCP client
resources = get_deployment_resources()

# Log model with resources for deployment
with mlflow.start_run():
    mlflow.pyfunc.log_model(
        artifact_path="agent",
        python_model=my_agent,
        resources=[
            DatabricksVectorSearchIndex(index_name=idx)
            for idx in resources.get("vector_search_indexes", [])
        ],
    )

Complete Example: Create Tools for Any Agent

"""
Create MCP tools that can be used with any agent framework.
"""
import asyncio
import os
from databricks_mcp import DatabricksMultiServerMCPClient, DatabricksMCPServer


async def create_mcp_tools():
    """Create LangChain-compatible tools from Databricks MCP servers."""

    # Define MCP servers to connect to
    servers = [
        # Vector Search for document retrieval
        DatabricksMCPServer(
            name="docs_search",
            server_url="/api/2.0/mcp/vector-search/main/docs"
        ),
        # Genie Space for data analysis
        DatabricksMCPServer(
            name="analytics",
            server_url="/api/2.0/mcp/genie/my_analytics_space"
        ),
        # UC Functions for custom operations
        DatabricksMCPServer(
            name="custom_functions",
            server_url="/api/2.0/mcp/functions/main/tools"
        ),
    ]

    async with DatabricksMultiServerMCPClient(servers=servers) as client:
        # Get all tools
        tools = await client.get_tools()

        # Get resources for deployment
        resources = client.get_databricks_resources()

        return tools, resources


def main():
    """Main function to create and display tools."""
    tools, resources = asyncio.run(create_mcp_tools())

    print(f"Created {len(tools)} MCP tools:")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description[:50]}...")

    print(f"\nDatabricks resources for deployment:")
    for resource_type, items in resources.items():
        print(f"  - {resource_type}: {items}")

    # Tools are now ready to use with any agent framework:
    # - LangGraph: graph.add_node("tools", ToolNode(tools))
    # - LangChain: agent = create_react_agent(llm, tools)
    # - Custom: for tool in tools: tool.invoke(input)

    return tools


if __name__ == "__main__":
    main()

Best Practices

1. Use Async Context Managers: Always use async with to ensure proper cleanup
2. Aggregate Tools Once: Create tools at startup, not per-request
3. Include Resources for Deployment: Use get_databricks_resources() when deploying to Model Serving
4. Use Meta Parameters: Fine-tune Vector Search with _meta for better results
5. OAuth for Production: Use OAuth authentication for service principals

References

• Databricks MCP Documentation: https://docs.databricks.com/aws/en/generative-ai/mcp
• databricks-mcp Package: https://pypi.org/project/databricks-mcp/
• databricks-langchain Package: https://pypi.org/project/databricks-langchain/
• langchain-mcp-adapters: https://pypi.org/project/langchain-mcp-adapters/
• MCP Protocol: https://modelcontextprotocol.io