Integration Guide

Integration Guide - Using Schema.org MCP in Your Applications

Overview

This guide covers how to integrate the Schema.org MCP server into various applications and platforms.

Table of Contents

  1. Claude Desktop
  2. Node.js Applications
  3. Python Applications
  4. Web Applications
  5. Command Line Tools
  6. Custom Integrations

Claude Desktop

Claude Desktop is the easiest way to use the Schema.org MCP server with Claude.

Setup

  1. Locate Configuration File

    macOS:

    ~/Library/Application Support/Claude/claude_desktop_config.json

    Linux:

    ~/.config/Claude/claude_desktop_config.json

    Windows:

    %APPDATA%\Claude\claude_desktop_config.json
  2. Add MCP Server Configuration

    {
      "mcpServers": {
        "schema-org": {
          "url": "https://json-ld-schema-mcp.serptransformer.com",
          "headers": {
           "Authorization": "Bearer sk_your_api_key"
         }
       }
     }
    }
  3. Restart Claude Desktop

    Close and reopen Claude Desktop to load the new configuration.

Usage in Claude

Once configured, you can ask Claude to use the schema.org tools:

Example Prompts:

"What schema types should I use for a blog post?"

"Show me all the properties available for the Organization schema type"

"Generate an example JSON-LD for a Recipe with these ingredients: flour, sugar, eggs"

"Search for schema types related to e-commerce products"

"What's the inheritance hierarchy for NewsArticle?"

Troubleshooting

Tools not appearing:

  • Verify configuration file syntax (valid JSON)
  • Check API key is correct
  • Restart Claude Desktop
  • Check server is running: curl https://json-ld-schema-mcp.serptransformer.com/health

Connection errors:

  • Verify HTTPS is used (not HTTP)
  • Check firewall allows outbound HTTPS
  • Verify DNS resolution: nslookup json-ld-schema-mcp.serptransformer.com

Node.js Applications

Installation

npm install @modelcontextprotocol/sdk

Basic Client

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js';
 
const apiKey = process.env.SCHEMA_ORG_API_KEY || 'sk_your_api_key';
 
async function main() {
  // Create transport
  const transport = new SSEClientTransport({
    url: 'https://json-ld-schema-mcp.serptransformer.com/sse',
    headers: {
      'Authorization': `Bearer ${apiKey}`,
    },
  });
 
  // Create client
  const client = new Client({
    name: 'my-schema-app',
    version: '1.0.0',
  }, {
    capabilities: {},
  });
 
  try {
    // Connect to server
    await client.connect(transport);
    console.log('Connected to Schema.org MCP server');
 
    // List available tools
    const tools = await client.listTools();
    console.log('Available tools:', tools.tools.map(t => t.name));
 
    // Call a tool
    const result = await client.callTool({
      name: 'get_schema_type',
      arguments: { typeName: 'Person' },
    });
    console.log('Person schema:', result);
 
  } catch (error) {
    console.error('Error:', error);
  } finally {
    await client.close();
  }
}
 
main();

Advanced Usage with Error Handling

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js';
 
class SchemaOrgClient {
  private client: Client;
  private transport: SSEClientTransport;
 
  constructor(apiKey: string) {
    this.transport = new SSEClientTransport({
      url: 'https://json-ld-schema-mcp.serptransformer.com/sse',
      headers: {
        'Authorization': `Bearer ${apiKey}`,
      },
    });
 
    this.client = new Client({
      name: 'schema-org-client',
      version: '1.0.0',
    }, {
      capabilities: {},
    });
  }
 
  async connect(): Promise<void> {
    await this.client.connect(this.transport);
  }
 
  async getSchemaType(typeName: string): Promise<any> {
    return this.client.callTool({
      name: 'get_schema_type',
      arguments: { typeName },
    });
  }
 
  async searchSchemas(query: string, limit: number = 10): Promise<any> {
    return this.client.callTool({
      name: 'search_schemas',
      arguments: { query, limit },
    });
  }
 
  async getTypeHierarchy(typeName: string): Promise<any> {
    return this.client.callTool({
      name: 'get_type_hierarchy',
      arguments: { typeName },
    });
  }
 
  async getTypeProperties(typeName: string, includeInherited: boolean = true): Promise<any> {
    return this.client.callTool({
      name: 'get_type_properties',
      arguments: { typeName, includeInherited },
    });
  }
 
  async generateExample(typeName: string, properties?: Record<string, any>): Promise<any> {
    return this.client.callTool({
      name: 'generate_example',
      arguments: { typeName, properties },
    });
  }
 
  async close(): Promise<void> {
    await this.client.close();
  }
}
 
// Usage
async function main() {
  const apiKey = process.env.SCHEMA_ORG_API_KEY || 'sk_your_api_key';
  const client = new SchemaOrgClient(apiKey);
 
  try {
    await client.connect();
 
    // Get schema type
    const person = await client.getSchemaType('Person');
    console.log('Person:', person);
 
    // Search schemas
    const articles = await client.searchSchemas('article', 5);
    console.log('Articles:', articles);
 
    // Get properties
    const props = await client.getTypeProperties('Organization');
    console.log('Organization properties:', props);
 
    // Generate example
    const example = await client.generateExample('Recipe', {
      name: 'Chocolate Cake',
      prepTime: 'PT30M',
    });
    console.log('Recipe example:', example);
 
  } finally {
    await client.close();
  }
}
 
main().catch(console.error);

Express.js Integration

import express from 'express';
import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js';
 
const app = express();
const apiKey = process.env.SCHEMA_ORG_API_KEY || 'sk_your_api_key';
 
let schemaClient: Client;
 
// Initialize client on startup
async function initializeClient() {
  const transport = new SSEClientTransport({
    url: 'https://json-ld-schema-mcp.serptransformer.com/sse',
    headers: {
      'Authorization': `Bearer ${apiKey}`,
    },
  });
 
  schemaClient = new Client({
    name: 'express-schema-app',
    version: '1.0.0',
  }, {
    capabilities: {},
  });
 
  await schemaClient.connect(transport);
}
 
// Routes
app.get('/api/schema/:type', async (req, res) => {
  try {
    const result = await schemaClient.callTool({
      name: 'get_schema_type',
      arguments: { typeName: req.params.type },
    });
    res.json(result);
  } catch (error) {
    res.status(400).json({ error: error.message });
  }
});
 
app.get('/api/search', async (req, res) => {
  try {
    const result = await schemaClient.callTool({
      name: 'search_schemas',
      arguments: {
        query: req.query.q as string,
        limit: parseInt(req.query.limit as string) || 10,
      },
    });
    res.json(result);
  } catch (error) {
    res.status(400).json({ error: error.message });
  }
});
 
app.get('/api/properties/:type', async (req, res) => {
  try {
    const result = await schemaClient.callTool({
      name: 'get_type_properties',
      arguments: {
        typeName: req.params.type,
        includeInherited: req.query.inherited !== 'false',
      },
    });
    res.json(result);
  } catch (error) {
    res.status(400).json({ error: error.message });
  }
});
 
// Start server
initializeClient().then(() => {
  app.listen(3001, () => {
    console.log('Server running on http://localhost:3001');
  });
}).catch(console.error);

Python Applications

Installation

pip install requests sseclient-py

Basic Client

import requests
import json
import os
 
class SchemaOrgMCP:
    def __init__(self, api_key=None):
        self.api_key = api_key or os.getenv('SCHEMA_ORG_API_KEY')
        self.base_url = "https://json-ld-schema-mcp.serptransformer.com"
        self.headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
 
    def health_check(self):
        """Check server health"""
        response = requests.get(f"{self.base_url}/health")
        return response.json()
 
    def get_schema_type(self, type_name):
        """Get schema type information"""
        message = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "get_schema_type",
                "arguments": {"typeName": type_name}
            }
        }
        response = requests.post(
            f"{self.base_url}/messages",
            headers=self.headers,
            json=message
        )
        return response.json()
 
    def search_schemas(self, query, limit=10):
        """Search for schema types"""
        message = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "search_schemas",
                "arguments": {"query": query, "limit": limit}
            }
        }
        response = requests.post(
            f"{self.base_url}/messages",
            headers=self.headers,
            json=message
        )
        return response.json()
 
    def get_type_hierarchy(self, type_name):
        """Get type hierarchy"""
        message = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "get_type_hierarchy",
                "arguments": {"typeName": type_name}
            }
        }
        response = requests.post(
            f"{self.base_url}/messages",
            headers=self.headers,
            json=message
        )
        return response.json()
 
    def get_type_properties(self, type_name, include_inherited=True):
        """Get type properties"""
        message = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "get_type_properties",
                "arguments": {
                    "typeName": type_name,
                    "includeInherited": include_inherited
                }
            }
        }
        response = requests.post(
            f"{self.base_url}/messages",
            headers=self.headers,
            json=message
        )
        return response.json()
 
    def generate_example(self, type_name, properties=None):
        """Generate example JSON-LD"""
        message = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": "generate_example",
                "arguments": {
                    "typeName": type_name,
                    "properties": properties or {}
                }
            }
        }
        response = requests.post(
            f"{self.base_url}/messages",
            headers=self.headers,
            json=message
        )
        return response.json()
 
# Usage
if __name__ == "__main__":
    mcp = SchemaOrgMCP()
 
    print("Health:", mcp.health_check())
    print("\nPerson Type:", mcp.get_schema_type("Person"))
    print("\nSearch Results:", mcp.search_schemas("article", 3))
    print("\nType Hierarchy:", mcp.get_type_hierarchy("NewsArticle"))
    print("\nProperties:", mcp.get_type_properties("Organization"))
    print("\nExample:", mcp.generate_example("Recipe", {
        "name": "Chocolate Cake",
        "prepTime": "PT30M"
    }))

Flask Integration

from flask import Flask, jsonify, request
import os
from schema_org_mcp import SchemaOrgMCP
 
app = Flask(__name__)
api_key = os.getenv('SCHEMA_ORG_API_KEY')
mcp = SchemaOrgMCP(api_key)
 
@app.route('/api/health', methods=['GET'])
def health():
    return jsonify(mcp.health_check())
 
@app.route('/api/schema/<type_name>', methods=['GET'])
def get_schema(type_name):
    try:
        result = mcp.get_schema_type(type_name)
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 400
 
@app.route('/api/search', methods=['GET'])
def search():
    try:
        query = request.args.get('q', '')
        limit = request.args.get('limit', 10, type=int)
        result = mcp.search_schemas(query, limit)
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 400
 
@app.route('/api/properties/<type_name>', methods=['GET'])
def get_properties(type_name):
    try:
        include_inherited = request.args.get('inherited', 'true').lower() == 'true'
        result = mcp.get_type_properties(type_name, include_inherited)
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 400
 
@app.route('/api/example/<type_name>', methods=['POST'])
def generate_example(type_name):
    try:
        properties = request.get_json() or {}
        result = mcp.generate_example(type_name, properties)
        return jsonify(result)
    except Exception as e:
        return jsonify({"error": str(e)}), 400
 
if __name__ == '__main__':
    app.run(debug=True, port=5000)

Web Applications

React Integration

import React, { useState, useEffect } from 'react';
 
interface SchemaType {
  name: string;
  description: string;
  url: string;
}
 
export function SchemaExplorer() {
  const [searchQuery, setSearchQuery] = useState('');
  const [results, setResults] = useState<SchemaType[]>([]);
  const [loading, setLoading] = useState(false);
  const [error, setError] = useState<string | null>(null);
 
  const apiKey = process.env.REACT_APP_SCHEMA_ORG_API_KEY;
 
  const search = async (query: string) => {
    if (!query.trim()) return;
 
    setLoading(true);
    setError(null);
 
    try {
      const response = await fetch(
        'https://json-ld-schema-mcp.serptransformer.com/messages',
        {
          method: 'POST',
          headers: {
            'Authorization': `Bearer ${apiKey}`,
            'Content-Type': 'application/json',
          },
          body: JSON.stringify({
            jsonrpc: '2.0',
            id: 1,
            method: 'tools/call',
            params: {
              name: 'search_schemas',
              arguments: { query, limit: 10 },
            },
          }),
        }
      );
 
      const data = await response.json();
      setResults(data.result?.content?.[0]?.text ? JSON.parse(data.result.content[0].text) : []);
    } catch (err) {
      setError(err instanceof Error ? err.message : 'Unknown error');
    } finally {
      setLoading(false);
    }
  };
 
  return (
    <div className="schema-explorer">
      <h1>Schema.org Explorer</h1>
      <div className="search-box">
        <input
          type="text"
          value={searchQuery}
          onChange={(e) => setSearchQuery(e.target.value)}
          placeholder="Search schema types..."
          onKeyPress={(e) => e.key === 'Enter' && search(searchQuery)}
        />
        <button onClick={() => search(searchQuery)}>Search</button>
      </div>
 
      {error && <div className="error">{error}</div>}
      {loading && <div className="loading">Loading...</div>}
 
      <div className="results">
        {results.map((result) => (
          <div key={result.name} className="result-card">
            <h3>{result.name}</h3>
            <p>{result.description}</p>
            <a href={result.url} target="_blank" rel="noopener noreferrer">
              View on schema.org
            </a>
          </div>
        ))}
      </div>
    </div>
  );
}

Command Line Tools

Bash Script

#!/bin/bash
 
API_KEY="${SCHEMA_ORG_API_KEY:-sk_your_api_key}"
BASE_URL="https://json-ld-schema-mcp.serptransformer.com"
 
# Colors for output
GREEN='\033[0;32m'
RED='\033[0;31m'
NC='\033[0m'
 
function health_check() {
    echo "Checking server health..."
    curl -s "$BASE_URL/health" | jq .
}
 
function get_schema() {
    local type_name=$1
    if [ -z "$type_name" ]; then
        echo "Usage: get_schema <type_name>"
        return 1
    fi
 
    curl -s -H "Authorization: Bearer $API_KEY" \
      -X POST \
      -H "Content-Type: application/json" \
      -d "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"get_schema_type\",\"arguments\":{\"typeName\":\"$type_name\"}}}" \
      "$BASE_URL/messages" | jq .
}
 
function search_schemas() {
    local query=$1
    local limit=${2:-10}
 
    curl -s -H "Authorization: Bearer $API_KEY" \
      -X POST \
      -H "Content-Type: application/json" \
      -d "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"tools/call\",\"params\":{\"name\":\"search_schemas\",\"arguments\":{\"query\":\"$query\",\"limit\":$limit}}}" \
      "$BASE_URL/messages" | jq .
}
 
# Main
case "${1:-help}" in
    health)
        health_check
        ;;
    get)
        get_schema "$2"
        ;;
    search)
        search_schemas "$2" "$3"
        ;;
    *)
        echo "Usage: $0 {health|get <type>|search <query> [limit]}"
        ;;
esac

Custom Integrations

Generic HTTP Client

For any language/framework, you can use HTTP requests:

1. Establish SSE connection: GET /sse
   Headers: Authorization: Bearer sk_your_api_key

2. Send messages: POST /messages
   Headers: Authorization: Bearer sk_your_api_key
   Body: JSON-RPC 2.0 message

3. Receive responses via SSE stream

Example Request/Response

Request:

curl -X POST \
  -H "Authorization: Bearer sk_your_api_key" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
      "name": "get_schema_type",
      "arguments": {"typeName": "Person"}
    }
  }' \
  https://json-ld-schema-mcp.serptransformer.com/messages

Response:

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "{\"name\":\"Person\",\"description\":\"A person (alive, dead, undead, or fictional).\",\"id\":\"https://schema.org/Person\",\"type\":\"rdfs:Class\",\"superTypes\":[{\"name\":\"Thing\",\"id\":\"https://schema.org/Thing\"}],\"url\":\"https://schema.org/Person\"}"
      }
    ]
  }
}

Best Practices

  1. Error Handling: Always handle connection errors and timeouts
  2. API Key Management: Store API keys in environment variables, never hardcode
  3. Caching: Cache schema.org data locally to reduce API calls
  4. Rate Limiting: Implement client-side rate limiting
  5. Logging: Log all API calls for debugging
  6. Timeouts: Set appropriate timeouts for HTTP requests
  7. Retries: Implement exponential backoff for retries

Support

For issues or questions: