MCP (Model Context Protocol): Membangun Integrasi Tools untuk AI

# MCP (Model Context Protocol): Membangun Integrasi Tools untuk AI Model Context Protocol (MCP) adalah protokol standar terbuka yang dikembangkan oleh Anthropic untuk menghubungkan asisten AI dengan...

By Ruby Abdullah · · tutorial
MCPModel Context ProtocolAnthropicTool UsePython

MCP (Model Context Protocol): Membangun Integrasi Tools untuk AI

Model Context Protocol (MCP) adalah protokol standar terbuka yang dikembangkan oleh Anthropic untuk menghubungkan asisten AI dengan sumber data dan tools eksternal. Bayangkan MCP sebagai "USB-C untuk AI" — satu standar universal yang memungkinkan AI model apapun terhubung dengan tools apapun tanpa perlu integrasi khusus untuk setiap kombinasi.

Dalam tutorial ini, kita akan mempelajari arsitektur MCP secara mendalam, membangun MCP server dan client dari nol, serta membuat contoh praktis berupa database query server yang memungkinkan AI assistant melakukan query PostgreSQL dengan aman.

Mengapa MCP Penting?

Sebelum MCP, setiap integrasi antara AI dan tools eksternal memerlukan implementasi khusus. Jika Anda memiliki 5 AI model dan 10 tools, Anda memerlukan 50 integrasi berbeda. MCP menyelesaikan masalah ini dengan menyediakan satu protokol standar.

Keuntungan utama MCP:

  • Standarisasi: Satu protokol untuk semua integrasi AI-tools
  • Keamanan: Kontrol akses granular terhadap data dan operasi
  • Modularitas: Server dan client dapat dikembangkan secara independen
  • Ekosistem: Komunitas yang berkembang dengan banyak server siap pakai
  • Vendor-agnostic: Tidak terikat pada satu provider AI tertentu

Arsitektur MCP

MCP menggunakan arsitektur client-server dengan tiga komponen utama:

1. Host

Host adalah aplikasi AI yang digunakan pengguna, seperti Claude Desktop, Claude Code, atau IDE dengan fitur AI. Host mengelola koneksi dan menyediakan antarmuka pengguna.

2. Client

Client adalah komponen di dalam host yang mengelola koneksi ke satu MCP server. Setiap client memiliki koneksi 1:1 dengan satu server. Client menangani negosiasi kapabilitas dan routing pesan.

3. Server

Server adalah program yang mengekspos resources, tools, dan prompts kepada client. Server bisa berjalan secara lokal atau remote dan berkomunikasi melalui transport layer yang didukung.

┌─────────────────────────────────────┐

│ HOST │

│ (Claude Desktop / Claude Code) │

│ │

│ ┌──────────┐ ┌──────────┐ │

│ │ Client 1 │ │ Client 2 │ │

│ └────┬─────┘ └────┬─────┘ │

└───────┼───────────────┼─────────────┘

│ │

┌────▼─────┐ ┌────▼─────┐

│ Server A │ │ Server B │

│ (DB) │ │ (Files) │

└──────────┘ └──────────┘

Primitif Utama MCP

MCP mendefinisikan tiga primitif utama yang dapat diekspos oleh server:

Resources

Resources adalah data yang dapat dibaca oleh AI, seperti file, database record, atau API response. Resources diidentifikasi dengan URI dan bersifat read-only.

Tools

Tools adalah fungsi yang dapat dipanggil oleh AI untuk melakukan aksi. Tools memiliki parameter yang didefinisikan dengan JSON Schema dan mengembalikan hasil dalam format terstruktur.

Prompts

Prompts adalah template yang dapat digunakan ulang untuk interaksi AI. Prompts membantu pengguna melakukan tugas umum dengan cara yang terstruktur.

Instalasi

Python SDK

# Install menggunakan pip

pip install mcp

Atau menggunakan uv (direkomendasikan)

uv add mcp

Untuk development

uv add "mcp[cli]"

TypeScript SDK

# Install menggunakan npm

npm install @modelcontextprotocol/sdk

Atau menggunakan pnpm

pnpm add @modelcontextprotocol/sdk

MCP Inspector (untuk testing)

# Jalankan MCP Inspector

npx @modelcontextprotocol/inspector

Membangun MCP Server dengan Python

Mari kita mulai dengan membangun MCP server sederhana yang mengekspos resources, tools, dan prompts.

Server Dasar

# server.py

from mcp.server.fastmcp import FastMCP

Inisialisasi server

mcp = FastMCP("Demo Server")

Definisikan resource

@mcp.resource("config://app")

def getconfig() -> str:

"""Mengembalikan konfigurasi aplikasi."""

return "App Configuration: debug=true, version=1.0"

Definisikan tool

@mcp.tool()

def calculatebmi(weightkg: float, heightm: float) -> str:

"""Menghitung Body Mass Index (BMI).

Args:

weightkg: Berat badan dalam kilogram

heightm: Tinggi badan dalam meter

"""

bmi = weightkg / (heightm * 2)

category = ""

if bmi < 18.5:

category = "Underweight"

elif bmi < 25:

category = "Normal"

elif bmi < 30:

category = "Overweight"

else:

category = "Obese"

return f"BMI: {bmi:.1f} ({category})"

Definisikan prompt

@mcp.prompt()

def reviewcode(code: str) -> str:

"""Prompt untuk review kode."""

return f"Tolong review kode berikut dan berikan saran perbaikan:\n\n{code}"

if name == "main":

mcp.run()

Menjalankan Server

# Menggunakan stdio transport (default)

python server.py

Menggunakan SSE transport

mcp run server.py --transport sse

Menggunakan streamable HTTP transport

mcp run server.py --transport streamable-http

Definisi Tool dengan Parameter

Tool adalah fitur paling powerful dalam MCP. Berikut cara mendefinisikan tool yang lebih kompleks:

from mcp.server.fastmcp import FastMCP

from typing import Optional

import json

mcp = FastMCP("Advanced Tools Server")

@mcp.tool()

def searchproducts(

query: str,

category: Optional[str] = None,

minprice: Optional[float] = None,

maxprice: Optional[float] = None,

limit: int = 10

) -> str:

"""Mencari produk berdasarkan query dan filter.

Args:

query: Kata kunci pencarian

category: Kategori produk (opsional)

minprice: Harga minimum (opsional)

maxprice: Harga maksimum (opsional)

limit: Jumlah hasil maksimum (default: 10)

"""

# Simulasi pencarian produk

results = {

"query": query,

"filters": {

"category": category,

"minprice": minprice,

"maxprice": maxprice

},

"limit": limit,

"results": [

{"name": "Produk A", "price": 100000, "category": "elektronik"},

{"name": "Produk B", "price": 250000, "category": "fashion"}

]

}

return json.dumps(results, indent=2, ensureascii=False)

@mcp.tool()

def createinvoice(

customername: str,

items: list[dict],

taxrate: float = 0.11

) -> str:

"""Membuat invoice baru.

Args:

customername: Nama pelanggan

items: Daftar item [{name, quantity, price}]

taxrate: Tarif pajak (default 11%)

"""

subtotal = sum(item["price"] item["quantity"] for item in items)

tax = subtotal taxrate

total = subtotal + tax

invoice = {

"customer": customername,

"items": items,

"subtotal": subtotal,

"tax": tax,

"total": total

}

return json.dumps(invoice, indent=2, ensureascii=False)

Membangun MCP Client

Berikut cara membangun MCP client yang terhubung ke server:

# client.py

import asyncio

from mcp import ClientSession, StdioServerParameters

from mcp.client.stdio import stdioclient

async def main():

# Konfigurasi koneksi ke server

serverparams = StdioServerParameters(

command="python",

args=["server.py"],

env=None

)

async with stdioclient(serverparams) as (read, write):

async with ClientSession(read, write) as session:

# Inisialisasi koneksi

await session.initialize()

# List semua tools yang tersedia

tools = await session.listtools()

print("Tools tersedia:")

for tool in tools.tools:

print(f" - {tool.name}: {tool.description}")

# List semua resources

resources = await session.listresources()

print("\nResources tersedia:")

for resource in resources.resources:

print(f" - {resource.uri}: {resource.name}")

# Panggil tool

result = await session.calltool(

"calculatebmi",

arguments={"weightkg": 70, "heightm": 1.75}

)

print(f"\nHasil BMI: {result.content[0].text}")

# Baca resource

resourcecontent = await session.readresource("config://app")

print(f"\nConfig: {resourcecontent.contents[0].text}")

if name == "main":

asyncio.run(main())

Transport Types

MCP mendukung beberapa jenis transport untuk komunikasi antara client dan server:

1. Stdio (Standard Input/Output)

Transport default yang paling sederhana. Client menjalankan server sebagai subprocess dan berkomunikasi melalui stdin/stdout.

# Client menggunakan stdio

serverparams = StdioServerParameters(

command="python",

args=["server.py"]

)

Kelebihan: Sederhana, tidak perlu konfigurasi jaringan Kekurangan: Hanya untuk koneksi lokal

2. SSE (Server-Sent Events)

Transport berbasis HTTP yang memungkinkan koneksi remote.

# Server dengan SSE transport

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("SSE Server")

Jalankan dengan SSE

mcp run server.py --transport sse --port 8080

# Client menggunakan SSE

from mcp.client.sse import sseclient

async with sseclient("http://localhost:8080/sse") as (read, write):

async with ClientSession(read, write) as session:

await session.initialize()

3. Streamable HTTP

Transport terbaru yang lebih efisien, mendukung koneksi stateless dan stateful.

# Server dengan streamable HTTP

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("HTTP Server", statelesshttp=True)

Jalankan dengan streamable HTTP

mcp run server.py --transport streamable-http --port 8080

Integrasi dengan Claude Desktop

Untuk menggunakan MCP server dengan Claude Desktop, edit file konfigurasi:

MacOS: ~/Library/Application Support/Claude/claudedesktopconfig.json Windows: %APPDATA%\Claude\claudedesktopconfig.json Linux: ~/.config/Claude/claudedesktopconfig.json
{

"mcpServers": {

"demo-server": {

"command": "python",

"args": ["/path/to/server.py"],

"env": {

"DATABASEURL": "postgresql://user:pass@localhost/mydb"

}

},

"another-server": {

"command": "npx",

"args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/documents"]

}

}

}

Setelah menyimpan konfigurasi, restart Claude Desktop. Anda akan melihat ikon tools (palu) di bagian bawah input chat yang menunjukkan tools yang tersedia.

Integrasi dengan Claude Code

Claude Code mendukung MCP server secara native. Anda dapat menambahkan server melalui CLI:

# Menambahkan MCP server ke project scope

claude mcp add db-server -s project -- python /path/to/dbserver.py

Menambahkan dengan environment variables

claude mcp add db-server -s project \

-e DATABASEURL=postgresql://user:pass@localhost/mydb \

-- python /path/to/dbserver.py

Menambahkan server SSE/HTTP remote

claude mcp add remote-server -s user --transport http \

http://api.example.com/mcp

List semua MCP server

claude mcp list

Menghapus MCP server

claude mcp remove db-server

Konfigurasi disimpan di file .mcp.json pada root project:

{

"mcpServers": {

"db-server": {

"command": "python",

"args": ["/path/to/dbserver.py"],

"env": {

"DATABASEURL": "postgresql://user:pass@localhost/mydb"

}

}

}

}

Testing dengan MCP Inspector

MCP Inspector adalah tool visual untuk testing dan debugging MCP server:

# Jalankan Inspector

npx @modelcontextprotocol/inspector

Inspector akan membuka browser di http://localhost:6274

Dengan Inspector, Anda dapat:

  • Melihat semua resources, tools, dan prompts yang diekspos server
  • Memanggil tools dengan parameter kustom
  • Membaca resources
  • Memeriksa respons server secara real-time
  • Debug masalah koneksi

Anda juga bisa menjalankan Inspector langsung dengan server:

# Langsung test server tertentu

npx @modelcontextprotocol/inspector python server.py

Test server dengan environment variables

DATABASEURL=postgresql://localhost/mydb \

npx @modelcontextprotocol/inspector python server.py

Pertimbangan Keamanan

Keamanan sangat penting saat membangun MCP server. Berikut praktik terbaik:

1. Validasi Input

import re

@mcp.tool()

def querydata(tablename: str, limit: int = 10) -> str:

"""Query data dari tabel tertentu."""

# Validasi nama tabel - hanya alfanumerik dan underscore

if not re.match(r'^[a-zA-Z][a-zA-Z0-9]

, tablename):

return "Error: Nama tabel tidak valid"

# Batasi limit

if limit < 1 or limit > 100:

return "Error: Limit harus antara 1 dan 100"

# Lanjutkan query...

2. Gunakan Parameterized Queries

Jangan pernah menggunakan string concatenation untuk SQL queries. Selalu gunakan parameterized queries untuk mencegah SQL injection.

3. Principle of Least Privilege

Berikan hanya akses minimum yang diperlukan. Jika server hanya perlu membaca data, gunakan database user dengan akses read-only.

4. Rate Limiting

Terapkan rate limiting untuk mencegah penyalahgunaan, terutama untuk server yang diekspos ke jaringan.

5. Audit Logging

Catat setiap operasi yang dilakukan melalui MCP server untuk keperluan audit dan debugging.

Contoh Praktis: Database Query MCP Server

Sekarang mari kita bangun contoh nyata — MCP server yang memungkinkan AI assistant melakukan query PostgreSQL dengan aman.

Persiapan

# Install dependencies

pip install mcp psycopg2-binary

Atau menggunakan uv

uv add mcp psycopg2-binary

Implementasi Server

# dbserver.py

import os

import json

import re

import psycopg2

from psycopg2.extras import RealDictCursor

from mcp.server.fastmcp import FastMCP

from contextlib import contextmanager

Inisialisasi MCP server

mcp = FastMCP("PostgreSQL Query Server")

Konfigurasi database

DATABASEURL = os.environ.get(

"DATABASEURL",

"postgresql://readonlyuser:password@localhost:5432/mydb"

)

Daftar tabel yang diizinkan

ALLOWEDTABLES = [

"products", "categories", "customers",

"orders", "orderitems"

]

Kolom sensitif yang harus di-mask

SENSITIVECOLUMNS = [

"password", "creditcard", "ssn",

"token", "secret", "apikey"

]

@contextmanager

def getconnection():

"""Context manager untuk koneksi database."""

conn = psycopg2.connect(DATABASEURL)

try:

yield conn

finally:

conn.close()

def sanitizeidentifier(name: str) -> str:

"""Validasi dan sanitasi SQL identifier."""

if not re.match(r'^[a-zA-Z][a-zA-Z0-9]

, name):

raise ValueError(f"Identifier tidak valid: {name}")

return name

def masksensitivedata(rows: list, columns: list) -> list:

"""Mask data sensitif dalam hasil query."""

maskedrows = []

for row in rows:

maskedrow = {}

for key, value in row.items():

if key.lower() in SENSITIVECOLUMNS:

maskedrow[key] = "MASKED"

else:

maskedrow[key] = value

maskedrows.append(maskedrow)

return maskedrows

Resources

@mcp.resource("schema://tables")

def listtables() -> str:

"""Daftar semua tabel yang dapat diakses."""

with getconnection() as conn:

with conn.cursor() as cur:

cur.execute("""

SELECT tablename,

pgsizepretty(

pgtotalrelationsize(

quoteident(tablename)

)

) as size

FROM informationschema.tables

WHERE tableschema = 'public'

AND tablename = ANY(%s)

ORDER BY tablename

""", (ALLOWEDTABLES,))

tables = cur.fetchall()

result = "Tabel yang tersedia:\n"

for tablename, size in tables:

result += f" - {tablename} ({size})\n"

return result

@mcp.resource("schema://tables/{tablename}")

def gettableschema(tablename: str) -> str:

"""Dapatkan schema dari tabel tertentu."""

tablename = sanitizeidentifier(tablename)

if tablename not in ALLOWEDTABLES:

return f"Error: Tabel '{tablename}' tidak diizinkan"

with getconnection() as conn:

with conn.cursor() as cur:

cur.execute("""

SELECT columnname, datatype,

isnullable, columndefault

FROM informationschema.columns

WHERE tableschema = 'public'

AND tablename = %s

ORDER BY ordinalposition

""", (tablename,))

columns = cur.fetchall()

result = f"Schema tabel '{tablename}':\n"

for colname, datatype, nullable, default in columns:

nullstr = "NULL" if nullable == "YES" else "NOT NULL"

defaultstr = f" DEFAULT {default}" if default else ""

result += f" - {colname}: {datatype} {nullstr}{defaultstr}\n"

return result

Tools

@mcp.tool()

def querytable(

tablename: str,

columns: list[str] | None = None,

whereclause: str | None = None,

orderby: str | None = None,

limit: int = 20

) -> str:

"""Query data dari tabel PostgreSQL dengan filter opsional.

Args:

tablename: Nama tabel yang akan di-query

columns: Daftar kolom yang akan ditampilkan (default: semua)

whereclause: Kondisi WHERE (contoh: "status = 'active'")

orderby: Kolom untuk pengurutan (contoh: "createdat DESC")

limit: Jumlah baris maksimum (default: 20, maks: 100)

"""

# Validasi tabel

tablename = sanitizeidentifier(tablename)

if tablename not in ALLOWEDTABLES:

return json.dumps({

"error": f"Tabel '{tablename}' tidak diizinkan. "

f"Tabel yang tersedia: {', '.join(ALLOWEDTABLES)}"

})

# Batasi limit

limit = min(max(1, limit), 100)

# Bangun query

if columns:

# Validasi setiap nama kolom

validatedcols = []

for col in columns:

validatedcols.append(sanitizeidentifier(col))

selectclause = ", ".join(validatedcols)

else:

selectclause = ""

query = f"SELECT {selectclause} FROM {tablename}"

params = []

# Tambahkan WHERE clause dengan hati-hati

if whereclause:

# Cegah SQL injection dasar

dangerouskeywords = [

"DROP", "DELETE", "UPDATE", "INSERT",

"ALTER", "EXEC", "EXECUTE", "--", ";"

]

upperwhere = whereclause.upper()

for keyword in dangerouskeywords:

if keyword in upperwhere:

return json.dumps({

"error": f"WHERE clause mengandung keyword "

f"berbahaya: {keyword}"

})

query += f" WHERE {whereclause}"

# Tambahkan ORDER BY

if orderby:

ordercol = sanitizeidentifier(

orderby.replace(" DESC", "")

.replace(" ASC", "")

.strip()

)

direction = "DESC" if "DESC" in orderby.upper() else "ASC"

query += f" ORDER BY {ordercol} {direction}"

query += f" LIMIT {limit}"

# Eksekusi query

with getconnection() as conn:

with conn.cursor(cursorfactory=RealDictCursor) as cur:

cur.execute(query)

rows = cur.fetchall()

# Mask data sensitif

rows = masksensitivedata(rows, columns or [])

# Konversi untuk serialisasi JSON

serializablerows = []

for row in rows:

serializablerow = {}

for key, value in row.items():

if hasattr(value, 'isoformat'):

serializablerow[key] = value.isoformat()

else:

serializablerow[key] = value

serializablerows.append(serializablerow)

return json.dumps({

"table": tablename,

"rowcount": len(serializablerows),

"data": serializablerows

}, indent=2, ensureascii=False, default=str)

@mcp.tool()

def gettablestats(tablename: str) -> str:

"""Dapatkan statistik dari tabel tertentu.

Args:

tablename: Nama tabel untuk statistik

"""

tablename = sanitizeidentifier(tablename)

if tablename not in ALLOWEDTABLES:

return json.dumps({

"error": f"Tabel '{tablename}' tidak diizinkan"

})

with getconnection() as conn:

with conn.cursor(cursorfactory=RealDictCursor) as cur:

# Hitung total baris

cur.execute(

f"SELECT COUNT() as total FROM {tablename}"

)

count = cur.fetchone()["total"]

# Dapatkan kolom numerik untuk statistik

cur.execute("""

SELECT columnname

FROM informationschema.columns

WHERE tableschema = 'public'

AND tablename = %s

AND datatype IN (

'integer', 'bigint', 'numeric',

'real', 'double precision'

)

""", (tablename,))

numericcols = [

row["columnname"] for row in cur.fetchall()

]

stats = {"table": tablename, "totalrows": count}

# Hitung statistik untuk kolom numerik

for col in numericcols:

if col.lower() in SENSITIVECOLUMNS:

continue

col = sanitizeidentifier(col)

cur.execute(f"""

SELECT

MIN({col}) as minval,

MAX({col}) as maxval,

AVG({col})::numeric(10,2) as avgval

FROM {tablename}

""")

colstats = cur.fetchone()

stats[f"{col}stats"] = {

"min": float(colstats["minval"])

if colstats["minval"] else None,

"max": float(colstats["maxval"])

if colstats["maxval"] else None,

"avg": float(colstats["avgval"])

if colstats["avgval"] else None

}

return json.dumps(stats, indent=2, ensureascii=False)

@mcp.tool()

def runaggregatequery(

tablename: str,

aggregatefunction: str,

column: str,

groupby: str | None = None,

whereclause: str | None = None

) -> str:

"""Jalankan query agregasi pada tabel.

Args:

tablename: Nama tabel

aggregatefunction: Fungsi agregasi (COUNT, SUM, AVG, MIN, MAX)

column: Kolom untuk agregasi (gunakan untuk COUNT)

groupby: Kolom untuk grouping (opsional)

whereclause: Kondisi WHERE (opsional)

"""

# Validasi

tablename = sanitizeidentifier(tablename)

if tablename not in ALLOWEDTABLES:

return json.dumps({

"error": f"Tabel '{tablename}' tidak diizinkan"

})

allowedfunctions = ["COUNT", "SUM", "AVG", "MIN", "MAX"]

aggfunc = aggregatefunction.upper()

if aggfunc not in allowedfunctions:

return json.dumps({

"error": f"Fungsi '{aggfunc}' tidak diizinkan. "

f"Gunakan: {', '.join(allowedfunctions)}"

})

# Bangun query

if column == "":

aggexpr = f"{aggfunc}()"

else:

column = sanitizeidentifier(column)

aggexpr = f"{aggfunc}({column})"

if groupby:

groupcol = sanitizeidentifier(groupby)

query = (

f"SELECT {groupcol}, {aggexpr} as result "

f"FROM {tablename}"

)

else:

query = f"SELECT {aggexpr} as result FROM {tablename}"

if whereclause:

dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "--", ";"]

for kw in dangerous:

if kw in whereclause.upper():

return json.dumps({

"error": f"WHERE clause tidak valid"

})

query += f" WHERE {whereclause}"

if groupby:

query += f" GROUP BY {groupcol} ORDER BY result DESC"

query += " LIMIT 50"

with getconnection() as conn:

with conn.cursor(cursorfactory=RealDictCursor) as cur:

cur.execute(query)

rows = cur.fetchall()

return json.dumps({

"querytype": f"{aggfunc}({column})",

"table": tablename,

"results": [dict(row) for row in rows]

}, indent=2, ensureascii=False, default=str)

Prompts

@mcp.prompt()

def analyzetable(tablename: str) -> str:

"""Prompt untuk analisis tabel database."""

return (

f"Tolong analisis tabel '{tablename}' di database. "

f"Langkah-langkah:\n"

f"1. Lihat schema tabel menggunakan resource schema://tables/{tablename}\n"

f"2. Dapatkan statistik tabel menggunakan tool gettablestats\n"

f"3. Query beberapa sample data menggunakan tool querytable\n"

f"4. Berikan insight tentang data yang ada\n"

f"5. Sarankan query yang berguna untuk analisis lebih lanjut"

)

@mcp.prompt()

def generatereport(tablename: str, metric: str) -> str:

"""Prompt untuk generate laporan dari data."""

return (

f"Buatkan laporan tentang {metric} dari tabel '{tablename}'. "

f"Gunakan tool yang tersedia untuk:\n"

f"1. Query data yang relevan\n"

f"2. Hitung agregasi yang diperlukan\n"

f"3. Format hasilnya dalam tabel markdown\n"

f"4. Berikan kesimpulan dan rekomendasi"

)

if name == "main":

mcp.run()

Setup Database untuk Testing

-- Buat user read-only

CREATE USER readonlyuser WITH PASSWORD 'password';

-- Buat database

CREATE DATABASE mydb;

-- Grant akses read-only

GRANT CONNECT ON DATABASE mydb TO readonlyuser;

GRANT USAGE ON SCHEMA public TO readonlyuser;

GRANT SELECT ON ALL TABLES IN SCHEMA public TO readonlyuser;

ALTER DEFAULT PRIVILEGES IN SCHEMA public

GRANT SELECT ON TABLES TO readonlyuser;

-- Buat tabel contoh

CREATE TABLE categories (

id SERIAL PRIMARY KEY,

name VARCHAR(100) NOT NULL,

description TEXT

);

CREATE TABLE products (

id SERIAL PRIMARY KEY,

name VARCHAR(200) NOT NULL,

categoryid INTEGER REFERENCES categories(id),

price NUMERIC(10, 2) NOT NULL,

stock INTEGER DEFAULT 0,

createdat TIMESTAMP DEFAULT CURRENTTIMESTAMP

);

CREATE TABLE customers (

id SERIAL PRIMARY KEY,

name VARCHAR(200) NOT NULL,

email VARCHAR(200) UNIQUE NOT NULL,

city VARCHAR(100),

createdat TIMESTAMP DEFAULT CURRENTTIMESTAMP

);

CREATE TABLE orders (

id SERIAL PRIMARY KEY,

customerid INTEGER REFERENCES customers(id),

totalamount NUMERIC(12, 2),

status VARCHAR(20) DEFAULT 'pending',

createdat TIMESTAMP DEFAULT CURRENTTIMESTAMP

);

CREATE TABLE orderitems (

id SERIAL PRIMARY KEY,

orderid INTEGER REFERENCES orders(id),

productid INTEGER REFERENCES products(id),

quantity INTEGER NOT NULL,

price NUMERIC(10, 2) NOT NULL

);

-- Insert data contoh

INSERT INTO categories (name, description) VALUES

('Elektronik', 'Perangkat elektronik dan gadget'),

('Fashion', 'Pakaian dan aksesoris'),

('Makanan', 'Makanan dan minuman');

INSERT INTO products (name, categoryid, price, stock) VALUES

('Laptop Pro 15', 1, 15000000, 50),

('Wireless Mouse', 1, 350000, 200),

('Kaos Polos', 2, 89000, 500),

('Kopi Arabika 250g', 3, 75000, 300);

Menghubungkan dengan Claude Desktop

{

"mcpServers": {

"postgres-query": {

"command": "python",

"args": ["/path/to/dbserver.py"],

"env": {

"DATABASEURL": "postgresql://readonlyuser:password@localhost:5432/mydb"

}

}

}

}

Menghubungkan dengan Claude Code

claude mcp add postgres-query -s project \

-e DATABASEURL=postgresql://readonlyuser:password@localhost:5432/mydb \

-- python /path/to/dbserver.py

Testing dengan MCP Inspector

DATABASEURL=postgresql://readonlyuser:password@localhost:5432/mydb \

npx @modelcontextprotocol/inspector python db_server.py

Best Practices

Berikut rangkuman best practices dalam membangun MCP server:

  • Deskripsi yang jelas: Tulis deskripsi tool dan parameter yang detail agar AI dapat memahami kapan dan bagaimana menggunakannya
  • Error handling yang baik: Kembalikan pesan error yang informatif, bukan stack trace
  • Validasi input selalu: Jangan pernah mempercayai input dari client tanpa validasi
  • Read-only by default: Mulai dengan akses read-only, tambahkan write access hanya jika diperlukan
  • Logging: Catat semua operasi untuk debugging dan audit
  • Environment variables: Simpan kredensial di environment variables, bukan hardcode
  • Testing: Gunakan MCP Inspector untuk testing menyeluruh sebelum deployment
  • Dokumentasi: Dokumentasikan setiap tool, resource, dan prompt yang diekspos
  • Kesimpulan

    MCP (Model Context Protocol) membuka paradigma baru dalam integrasi AI dengan tools dan data. Dengan membangun MCP server, Anda memberikan AI assistant akses terkontrol ke sistem Anda, memungkinkan otomasi yang lebih cerdas dan produktif.

    Dalam tutorial ini, kita telah mempelajari:

    Langkah selanjutnya yang bisa Anda eksplorasi:

    Selamat membangun integrasi AI yang powerful dan aman dengan MCP!

    Artikel Terkait

    PaddleOCR: Ekstraksi Teks dari Gambar dan Dokumen dengan Akurasi Tinggi

    PaddleOCR: Ekstraksi Teks dari Gambar dan Dokumen dengan Akurasi Tinggi Halo temen-temen, kali ini kita bahas salah satu...

    DSPy: Berhenti Ngoprek Prompt Manual, Biarkan Compiler yang Optimasi

    DSPy: Berhenti Ngoprek Prompt Manual, Biarkan Compiler yang Optimasi Halo temen-temen, kali ini aku mau ngenalin satu li...

    Hamilton: Bikin Dataflow dan Pipeline ML yang Rapi dan Gampang Dirawat di Python

    Hamilton: Bikin Dataflow dan Pipeline ML yang Rapi dan Gampang Dirawat di Python Halo temen-temen, ketemu lagi sama aku ...

    Inngest: Bikin Durable Functions dan Workflow Event-Driven Tanpa Ribet Ngurus Queue

    Inngest: Bikin Durable Functions dan Workflow Event-Driven Tanpa Ribet Ngurus Queue Temen-temen, kalau kamu pernah bangu...