Connecting Agent Frameworks to OpenMetadata MCP

This section also includes guides for Claude Desktop, Cursor, VS Code, and Goose. Each of those assumes a person runs the client interactively and can complete a browser sign-in. This guide covers the opposite case: a programmatic agent framework such as CrewAI, LangChain, the OpenAI Agents SDK, or your own orchestration code. It calls the MCP server directly over HTTP with no one present.

When to Use This Guide

Use this guide if:

You’re wiring an agent framework directly to the MCP endpoint, not through one of the interactive clients covered elsewhere in this section.
Your process runs unattended, on a schedule, or as part of a larger pipeline. No one is available to complete an OAuth browser login.
You need a service identity whose access is separate from any individual’s account.

1. Authenticate with a Bot Token

OAuth 2.0 (see OAuth 2.0 Authentication) requires a person to sign in through a browser, so it doesn’t fit an unattended agent. Bots also can’t generate a Personal Access Token — OpenMetadata rejects that request. Use a Bot’s JSON Web Token (JWT) instead:

Follow How to Set Up Bots to create a Bot and assign it the role-based access policy your agent needs.
On the Bot’s details page, click Generate New Token to issue its JWT, following Bot Token.
Store the token in an environment variable or secrets manager, never in source control.

Every request to the MCP endpoint carries this token in the Authorization header:

Authorization: Bearer <your-bot-token>

2. Set Up a Raw JSON-RPC Session

Before your first tool call, initialize the connection and confirm it:

// 1. POST {OMURL}/mcp
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {},
    "clientInfo": {
      "name": "my-agent",
      "version": "1.0.0"
    }
  }
}

// 2. POST {OMURL}/mcp, a notification with no "id" and no response body
{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

From here, call any tool with tools/call:

// POST {OMURL}/mcp
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "search_metadata",
    "arguments": {
      "query": "customer orders",
      "entityType": "table"
    }
  }
}

See the MCP Server Connection Guide for the full protocol reference, including headers, tools/list, and prompts. See the MCP Tools Reference for every available tool.

3. Wrap the Session in LangChain

The following snippet wraps the raw session shown earlier as a LangChain StructuredTool. It’s illustrative, not a maintained SDK: A full LangChain or CrewAI MCP adapter, if one becomes available, would replace hand-rolled HTTP calls like these. It targets Python 3.10+ because the code uses the X | None type-hint syntax introduced in that release.

import os
import httpx
from pydantic import BaseModel, Field
from langchain_core.tools import StructuredTool


class MCPSession:
    def __init__(self, base_url: str, bot_token: str):
        self.endpoint = base_url.rstrip("/") + "/mcp"
        self.headers = {
            "Authorization": f"Bearer {bot_token}",
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
        }
        self.client = httpx.Client(timeout=30)
        self._next_id = 1
        self._initialize()

    def _request(self, method: str, params: dict | None = None, notify: bool = False) -> dict | None:
        payload = {"jsonrpc": "2.0", "method": method}
        if not notify:
            payload["id"] = self._next_id
            self._next_id += 1
        if params is not None:
            payload["params"] = params
        response = self.client.post(self.endpoint, headers=self.headers, json=payload)
        response.raise_for_status()
        return response.json() if response.content else None

    def _initialize(self) -> None:
        self._request(
            "initialize",
            {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {"name": "langchain-agent", "version": "1.0.0"},
            },
        )
        self._request("notifications/initialized", notify=True)

    def call_tool(self, name: str, arguments: dict) -> str:
        response = self._request("tools/call", {"name": name, "arguments": arguments})
        assert response is not None, "tools/call must not be sent as a notification"
        result = response["result"]
        text = result["content"][0]["text"]
        if result.get("isError"):
            raise RuntimeError(f"{name} failed: {text}")
        return text


session = MCPSession(
    base_url=os.environ["OM_SERVER_URL"],
    bot_token=os.environ["OM_BOT_TOKEN"],
)


class SearchMetadataInput(BaseModel):
    query: str = Field(description="Keywords to search for")
    entity_type: str | None = Field(default=None, description="Optional entity type filter, e.g. 'table'")


def search_metadata(query: str, entity_type: str | None = None) -> str:
    arguments = {"query": query}
    if entity_type:
        arguments["entityType"] = entity_type
    return session.call_tool("search_metadata", arguments)


search_metadata_tool = StructuredTool.from_function(
    func=search_metadata,
    name="search_metadata",
    description="Search the OpenMetadata catalog by keyword.",
    args_schema=SearchMetadataInput,
)

The same pattern works for other tools, such as get_entity_details and get_entity_lineage. Wrap each one the same way: one small function per tool, each calling session.call_tool with that tool’s name and arguments.

4. Check isError on Every Response

Because there’s no human watching for a stuck spinner, a headless caller must explicitly check result.isError on every tool call rather than assuming a 200-shaped response succeeded. See Tool Execution Errors for the full envelope shape and a worked example.

​Connecting Agent Frameworks to OpenMetadata MCP

​When to Use This Guide

​1. Authenticate with a Bot Token

​2. Set Up a Raw JSON-RPC Session

​3. Wrap the Session in LangChain

​4. Check isError on Every Response