Developers

Examples

Small, complete programs you can copy. They read the key and the organization ID from environment variables and use nothing beyond the standard tools of each language. Our release pipeline runs every one of them, unchanged, against a test copy of C1K before each deployment.

REST examples need a key with the scopes links:read, links:write, analytics:read and events:read. MCP examples need an MCP token (Integrations, a key Used for: MCP) with links:read and links:write.

export C1K_ORG_ID='org_...'
export C1K_API_KEY='c1k_rest_...'    # for api.*
export C1K_MCP_TOKEN='c1k_mcp_...'   # for mcp.*

#!/usr/bin/env bash
# C1K REST API with curl (API-09). Set these first; never paste a key into a script.
#   export C1K_API_KEY='c1k_rest_...'      # from the Integrations page
#   export C1K_ORG_ID='org_...'            # shown next to the API base URL
# Needs the scopes links:read, links:write, analytics:read and events:read.
set -euo pipefail
: "${C1K_API_KEY:?set C1K_API_KEY}" "${C1K_ORG_ID:?set C1K_ORG_ID}"
BASE="https://c1k.me/api/v1/orgs/$C1K_ORG_ID"
# --fail-with-body stops the script on an error and still prints the error JSON.
C1K=(curl -sS --fail-with-body -H "Authorization: Bearer $C1K_API_KEY")

# Who am I, and what may this key do?
"${C1K[@]}" "$BASE/me"; echo

# Create a link. The Idempotency-Key makes a retry after a lost response safe:
# the same key with the same body returns the first result instead of a duplicate.
CREATED=$("${C1K[@]}" -H 'Content-Type: application/json' \
  -H "Idempotency-Key: example-$(date -u +%Y%m%d%H%M%S)-$RANDOM" \
  -d '{"destination_url":"https://venture.example/launch","title":"Launch page","utm":{"source":"newsletter","medium":"email"}}' \
  "$BASE/links")
echo "$CREATED"
LINK_ID=$(printf '%s' "$CREATED" | sed -n 's/.*"id":"\(lnk_[0-9A-Za-z]*\)".*/\1/p')

# Edit it. If-Match carries the current version, here 1 (also sent as the ETag header).
"${C1K[@]}" -X PATCH -H 'Content-Type: application/json' -H 'If-Match: "1"' \
  -d '{"destination_url":"https://venture.example/launch-v2"}' \
  "$BASE/links/$LINK_ID"; echo

# List links, newest first; pass meta.next_cursor as ?cursor= for the next page.
"${C1K[@]}" "$BASE/links?limit=25"; echo

# Organization analytics for the last seven UTC days, or an explicit range (to is exclusive).
"${C1K[@]}" "$BASE/analytics"; echo
"${C1K[@]}" "$BASE/analytics?from=$(date -u -d '6 days ago' +%F 2>/dev/null || date -u -v-6d +%F)&to=$(date -u -d tomorrow +%F 2>/dev/null || date -u -v+1d +%F)"; echo

# Poll the change feed. Keep meta.next_cursor and pass it back next time.
# A 410 cursor_expired means: take a fresh snapshot, then poll without a cursor.
"${C1K[@]}" "$BASE/events"; echo

// C1K REST API with fetch (Node.js 18 or newer, or a browser-less runtime). API-09.
//   C1K_API_KEY=c1k_rest_... C1K_ORG_ID=org_... node docs/examples/api.mjs
const key = process.env.C1K_API_KEY;
const org = process.env.C1K_ORG_ID;
if (!key || !org) throw new Error('Set C1K_API_KEY and C1K_ORG_ID');
const base = `https://c1k.me/api/v1/orgs/${org}`;

async function call(method, path, { body, headers = {} } = {}) {
  const res = await fetch(base + path, {
    method,
    headers: {
      Authorization: `Bearer ${key}`,
      ...(body ? { 'Content-Type': 'application/json' } : {}),
      ...headers,
    },
    body: body ? JSON.stringify(body) : undefined,
  });
  const payload = await res.json();
  if (!res.ok) {
    const e = payload.error;
    const retry = res.headers.get('Retry-After');
    throw new Error(`${res.status} ${e.code}: ${e.message}${retry ? ` (retry after ${retry}s)` : ''} [${e.request_id}]`);
  }
  return { payload, etag: res.headers.get('ETag') };
}

// Create with an idempotency key, so a retry after a lost response cannot duplicate the link.
const idem = `example-${crypto.randomUUID()}`;
const { payload: created, etag } = await call('POST', '/links', {
  body: { destination_url: 'https://venture.example/launch', title: 'Launch page', utm: { source: 'newsletter', medium: 'email' } },
  headers: { 'Idempotency-Key': idem },
});
console.log('created', created.data.short_url, 'version', created.data.version);

// Edit using the version from the ETag (If-Match, API-07).
const { payload: edited } = await call('PATCH', `/links/${created.data.id}`, {
  body: { title: 'Launch page (v2)' },
  headers: { 'If-Match': etag },
});
console.log('edited to version', edited.data.version);

// List every page.
let cursor = null;
do {
  const { payload } = await call('GET', `/links?limit=100${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ''}`);
  for (const link of payload.data) console.log(link.short_url, '->', link.destination_url);
  cursor = payload.meta.next_cursor;
} while (cursor);

// Analytics of the organization and of one link.
const { payload: stats } = await call('GET', '/analytics');
console.log('recorded requests', stats.data.totals.total, `(${stats.data.unique_estimate.label}: ${stats.data.unique_estimate.value})`);

// Poll the change feed once; store meta.next_cursor for the next poll.
const { payload: feed } = await call('GET', '/events');
for (const event of feed.data) console.log(event.id, event.time, event.action, event.target.id);
console.log('next cursor', feed.meta.next_cursor);

"""C1K REST API with the Python standard library only (API-09).

    C1K_API_KEY=c1k_rest_... C1K_ORG_ID=org_... python3 docs/examples/api.py
"""
import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request
import uuid

KEY = os.environ.get("C1K_API_KEY")
ORG = os.environ.get("C1K_ORG_ID")
if not KEY or not ORG:
    sys.exit("Set C1K_API_KEY and C1K_ORG_ID")
BASE = f"https://c1k.me/api/v1/orgs/{ORG}"


def call(method, path, body=None, headers=None):
    data = json.dumps(body).encode("utf-8") if body is not None else None
    req = urllib.request.Request(BASE + path, data=data, method=method)
    req.add_header("Authorization", f"Bearer {KEY}")
    if data is not None:
        req.add_header("Content-Type", "application/json")
    for name, value in (headers or {}).items():
        req.add_header(name, value)
    try:
        with urllib.request.urlopen(req, timeout=15) as res:
            return json.load(res), res.headers.get("ETag")
    except urllib.error.HTTPError as err:
        error = json.load(err).get("error", {})
        retry = err.headers.get("Retry-After")
        hint = f" (retry after {retry}s)" if retry else ""
        raise SystemExit(f"{err.code} {error.get('code')}: {error.get('message')}{hint} [{error.get('request_id')}]")


# Create with an idempotency key so a retry after a lost response cannot duplicate it.
created, etag = call(
    "POST",
    "/links",
    {"destination_url": "https://venture.example/launch", "title": "Launch page"},
    {"Idempotency-Key": f"example-{uuid.uuid4()}"},
)
link = created["data"]
print("created", link["short_url"], "version", link["version"])

# Edit with If-Match carrying the current version (API-07).
edited, _ = call("PATCH", f"/links/{link['id']}", {"title": "Launch page (v2)"}, {"If-Match": etag})
print("edited to version", edited["data"]["version"])

# List all pages.
cursor = None
while True:
    query = "?limit=100" + (f"&cursor={urllib.parse.quote(cursor)}" if cursor else "")
    page, _ = call("GET", "/links" + query)
    for item in page["data"]:
        print(item["short_url"], "->", item["destination_url"])
    cursor = page["meta"]["next_cursor"]
    if not cursor:
        break

# Analytics and one feed poll; keep meta.next_cursor for the next poll.
stats, _ = call("GET", "/analytics")
print("recorded requests", stats["data"]["totals"]["total"])
feed, _ = call("GET", "/events")
for event in feed["data"]:
    print(event["id"], event["time"], event["action"])
print("next cursor", feed["meta"]["next_cursor"])

<?php
// C1K REST API with plain PHP 7.0 or newer: no Composer, no curl extension,
// only json and openssl, which every PHP host has (API-09).
//   C1K_API_KEY=c1k_rest_... C1K_ORG_ID=org_... php docs/examples/api.php
$key = getenv('C1K_API_KEY');
$org = getenv('C1K_ORG_ID');
if (!$key || !$org) {
    fwrite(STDERR, "Set C1K_API_KEY and C1K_ORG_ID\n");
    exit(1);
}
define('C1K_BASE', 'https://c1k.me/api/v1/orgs/' . $org);
define('C1K_KEY', $key);

/** One API call. Returns array(payload, etag); throws with the error code and request ID. */
function c1k($method, $path, $body = null, array $headers = array())
{
    $lines = array('Authorization: Bearer ' . C1K_KEY, 'Accept: application/json');
    if ($body !== null) {
        $lines[] = 'Content-Type: application/json';
    }
    foreach ($headers as $name => $value) {
        $lines[] = $name . ': ' . $value;
    }
    $context = stream_context_create(array('http' => array(
        'method' => $method,
        'header' => implode("\r\n", $lines),
        'content' => $body === null ? '' : json_encode($body),
        'ignore_errors' => true,
        'timeout' => 15,
    )));
    $raw = file_get_contents(C1K_BASE . $path, false, $context);
    $status = 0;
    $etag = null;
    $retry = null;
    foreach (isset($http_response_header) ? $http_response_header : array() as $line) {
        if (preg_match('#^HTTP/\S+ (\d{3})#', $line, $m)) {
            $status = (int) $m[1];
        } elseif (stripos($line, 'ETag:') === 0) {
            $etag = trim(substr($line, 5));
        } elseif (stripos($line, 'Retry-After:') === 0) {
            $retry = trim(substr($line, 12));
        }
    }
    $payload = json_decode((string) $raw, true);
    if ($status < 200 || $status >= 300) {
        $e = isset($payload['error']) ? $payload['error'] : array('code' => 'network', 'message' => 'No answer from C1K.', 'request_id' => '-');
        throw new RuntimeException($status . ' ' . $e['code'] . ': ' . $e['message'] . ($retry !== null ? ' (retry after ' . $retry . 's)' : '') . ' [' . $e['request_id'] . ']');
    }
    return array($payload, $etag);
}

// Create with an idempotency key, so a retry after a lost response cannot duplicate the link.
list($created, $etag) = c1k('POST', '/links', array(
    'destination_url' => 'https://venture.example/launch',
    'title' => 'Launch page',
    'utm' => array('source' => 'newsletter', 'medium' => 'email'),
), array('Idempotency-Key' => 'example-' . bin2hex(random_bytes(8))));
$link = $created['data'];
echo 'created ', $link['short_url'], ' version ', $link['version'], "\n";

// Edit with If-Match carrying the current version from the ETag (API-07).
list($edited) = c1k('PATCH', '/links/' . $link['id'], array('title' => 'Launch page (v2)'), array('If-Match' => $etag));
echo 'edited to version ', $edited['data']['version'], "\n";

// List every page.
$cursor = null;
do {
    list($page) = c1k('GET', '/links?limit=100' . ($cursor !== null ? '&cursor=' . rawurlencode($cursor) : ''));
    foreach ($page['data'] as $item) {
        echo $item['short_url'], ' -> ', $item['destination_url'], "\n";
    }
    $cursor = $page['meta']['next_cursor'];
} while ($cursor !== null);

// Analytics, then one poll of the change feed; keep meta.next_cursor for the next poll.
list($stats) = c1k('GET', '/analytics');
echo 'recorded requests ', $stats['data']['totals']['total'], "\n";
list($feed) = c1k('GET', '/events');
foreach ($feed['data'] as $event) {
    echo $event['id'], ' ', $event['time'], ' ', $event['action'], "\n";
}
echo 'next cursor ', var_export($feed['meta']['next_cursor'], true), "\n";

#!/usr/bin/env bash
# C1K MCP endpoint with plain curl (MCP-01): one JSON-RPC message per POST,
# JSON back, no session. Useful to see what an MCP client does under the hood.
#   export C1K_MCP_TOKEN='c1k_mcp_...'   # Integrations page, a key "Used for: MCP"
#   export C1K_ORG_ID='org_...'
set -euo pipefail
: "${C1K_MCP_TOKEN:?set C1K_MCP_TOKEN}" "${C1K_ORG_ID:?set C1K_ORG_ID}"
URL="https://c1k.me/mcp/$C1K_ORG_ID"
HEADERS=(-H "Authorization: Bearer $C1K_MCP_TOKEN" -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream')

# After initialize, every request names the negotiated protocol version.
rpc() { curl -sS --fail-with-body "${HEADERS[@]}" -H 'MCP-Protocol-Version: 2025-11-25' -d "$1" "$URL"; echo; }

# 1. Initialize. The answer names the protocol version, 2025-11-25.
curl -sS --fail-with-body "${HEADERS[@]}" \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"curl-example","version":"1.0.0"}}}' \
  "$URL"; echo

# 2. Confirm initialization (answers 202 without a body).
rpc '{"jsonrpc":"2.0","method":"notifications/initialized"}'

# 3. The tools this token's scopes allow.
rpc '{"jsonrpc":"2.0","id":2,"method":"tools/list"}'

# 4. Create a link. Repeating the same idempotency_key returns the first result.
rpc "{\"jsonrpc\":\"2.0\",\"id\":3,\"method\":\"tools/call\",\"params\":{\"name\":\"create_link\",\"arguments\":{\"destination_url\":\"https://venture.example/launch\",\"idempotency_key\":\"mcp-example-$(date -u +%Y%m%d%H%M%S)\"}}}"

# 5. The ten newest links.
rpc '{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"list_links","arguments":{"limit":10}}}'

"""C1K MCP endpoint with the Python standard library only (MCP-01, MCP-02).

    C1K_MCP_TOKEN=c1k_mcp_... C1K_ORG_ID=org_... python3 docs/examples/mcp.py

Each call is one JSON-RPC message in one POST; the server keeps no session.
"""
import itertools
import json
import os
import sys
import time
import urllib.error
import urllib.request

TOKEN = os.environ.get("C1K_MCP_TOKEN")
ORG = os.environ.get("C1K_ORG_ID")
if not TOKEN or not ORG:
    sys.exit("Set C1K_MCP_TOKEN and C1K_ORG_ID")
URL = f"https://c1k.me/mcp/{ORG}"
VERSION = "2025-11-25"
ids = itertools.count(1)


def send(method, params=None, notification=False, initialized=True):
    message = {"jsonrpc": "2.0", "method": method}
    if params is not None:
        message["params"] = params
    if not notification:
        message["id"] = next(ids)
    req = urllib.request.Request(URL, data=json.dumps(message).encode("utf-8"), method="POST")
    req.add_header("Authorization", f"Bearer {TOKEN}")
    req.add_header("Content-Type", "application/json")
    req.add_header("Accept", "application/json, text/event-stream")
    if initialized:
        req.add_header("MCP-Protocol-Version", VERSION)
    try:
        with urllib.request.urlopen(req, timeout=15) as res:
            body = res.read()
    except urllib.error.HTTPError as err:
        raise SystemExit(f"HTTP {err.code}: {err.read().decode('utf-8', 'replace')}")
    if notification:
        return None
    reply = json.loads(body)
    if "error" in reply:
        raise SystemExit(f"JSON-RPC error {reply['error']['code']}: {reply['error']['message']}")
    return reply["result"]


def call_tool(name, arguments):
    result = send("tools/call", {"name": name, "arguments": arguments})
    if result.get("isError"):
        raise SystemExit(f"{name} failed: {result['content'][0]['text']}")
    return result["structuredContent"]


info = send("initialize", {"protocolVersion": VERSION, "capabilities": {}, "clientInfo": {"name": "python-example", "version": "1.0.0"}}, initialized=False)
print("connected to", info["serverInfo"]["title"], "protocol", info["protocolVersion"])
send("notifications/initialized", notification=True)

tools = send("tools/list")["tools"]
print("tools:", ", ".join(tool["name"] for tool in tools))

created = call_tool("create_link", {"destination_url": "https://venture.example/launch", "idempotency_key": f"mcp-example-{int(time.time())}"})
print("created", created["link"]["short_url"])

for link in call_tool("list_links", {"limit": 10})["links"]:
    print(link["short_url"], "->", link["destination_url"])

// C1K MCP endpoint with the official MCP TypeScript SDK, the way desktop
// agents and IDEs connect (MCP-01 to MCP-03).
//   npm install @modelcontextprotocol/sdk
//   C1K_MCP_TOKEN=c1k_mcp_... C1K_ORG_ID=org_... node mcp.mjs
import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';

const token = process.env.C1K_MCP_TOKEN;
const org = process.env.C1K_ORG_ID;
if (!token || !org) throw new Error('Set C1K_MCP_TOKEN and C1K_ORG_ID');

const client = new Client({ name: 'c1k-example', version: '1.0.0' });
await client.connect(new StreamableHTTPClientTransport(new URL(`https://c1k.me/mcp/${org}`), {
  requestInit: { headers: { Authorization: `Bearer ${token}` } },
}));
console.log('connected to', client.getServerVersion().title);

const { tools } = await client.listTools();
console.log('tools:', tools.map((tool) => tool.name).join(', '));

// The SDK checks structuredContent against each tool's outputSchema.
const created = await client.callTool({
  name: 'create_link',
  arguments: { destination_url: 'https://venture.example/launch', idempotency_key: `mcp-example-${Date.now()}` },
});
if (created.isError) throw new Error(created.content[0].text);
console.log('created', created.structuredContent.link.short_url);

const listed = await client.callTool({ name: 'list_links', arguments: { limit: 10 } });
for (const link of listed.structuredContent.links) console.log(link.short_url, '->', link.destination_url);

await client.close();