Appendix

This section contains reference configurations, scripts, and examples used across the documentation.

These can be used as supporting material while following the steps described in the main sections.

MCP Client Script (Python)

import argparse
import asyncio
import json
from fastmcp import Client
from fastmcp.client.auth import BearerAuth

async def main():
    parser = argparse.ArgumentParser(description="Generic MCP Test Client")
    
    parser.add_argument("--mcp_server", required=True, help="URL (http/sse) or local server python script path(stdio)")
    parser.add_argument("--mcp_auth_token", help="Optional bearer token for authentication")

    subparsers = parser.add_subparsers(dest="action_cmd", required=True)

    tool_parser = subparsers.add_parser("tools")
    tool_parser.add_argument("--list", action="store_true")
    tool_parser.add_argument("--tool_cmd")
    tool_parser.add_argument("--tool_args", default="{}")

    res_parser = subparsers.add_parser("resources")
    res_parser.add_argument("--list", action="store_true")
    res_parser.add_argument("--resource_uri")

    prompt_parser = subparsers.add_parser("prompts")
    prompt_parser.add_argument("--list", action="store_true")
    prompt_parser.add_argument("--prompt_name")
    prompt_parser.add_argument("--prompt_args", default="{}")

    args = parser.parse_args()

    if args.mcp_server.startswith(("http://", "https://")):
        transport_type = "sse" if "/sse" in args.mcp_server else "http"
        server_config = {
            "transport": transport_type,
            "url": args.mcp_server
        }
        if args.mcp_auth_token:
            server_config["auth"] = BearerAuth(token=f"{args.mcp_auth_token}")
    else:
        server_config = {
            "transport": "stdio",
            "command": "node" if args.mcp_server.endswith(".js") else "python",
            "args": [args.mcp_server]
        }

    config = {"mcpServers": {"target_server": server_config}}
    client = Client(config)
    
    async with client:
        if args.action_cmd == "tools":
            if args.list:
                tools = await client.list_tools()
                print(f"Available Tools: {tools}")
            elif args.tool_cmd:
                result = await client.call_tool(args.tool_cmd, json.loads(args.tool_args))
                print(f"Tool Result: {result}")

        elif args.action_cmd == "resources":
            if args.list:
                resources = await client.list_resources()
                print(f"Available Resources: {resources}")
            elif args.resource_uri:
                print(await client.read_resource(args.resource_uri))

        elif args.action_cmd == "prompts":
            if args.list:
                prompts = await client.list_prompts()
                print(f"Available Prompts: {prompts}")
            elif args.prompt_name:
                messages = await client.get_prompt(args.prompt_name, json.loads(args.prompt_args))
                print(messages)

if __name__ == "__main__":
    asyncio.run(main())

Source: MCP Sensor Observability

---

MCP File Server (HTTP) — Python

from fastmcp import FastMCP
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
import os

verifier = StaticTokenVerifier(tokens={"dev-token": {"client_id": "dev-client",
                                                     "scopes": ["read", "write"]}},
                               required_scopes=["read"])

mcp = FastMCP("FileServer", auth=verifier)
WORKING_DIR = os.path.abspath("./mcp-data")

os.makedirs(WORKING_DIR, exist_ok=True)

@mcp.resource("config://app-settings")
def get_config() -> str:
    return "Mode: Development; Version: 1.0.0; Allowed-Extensions: .txt, .md"

@mcp.resource("folder://explorer")
def list_files() -> str:
    files = os.listdir(WORKING_DIR)
    return "\n".join(files) if files else "The directory is empty."

@mcp.tool()
def write_file(filename: str, content: str) -> str:
    path = os.path.join(WORKING_DIR, filename)
    with open(path, "w") as f:
        f.write(content)
    return f"File '{filename}' written successfully."

@mcp.tool()
def remove_file(filename: str) -> str:
    file_path = os.path.join(WORKING_DIR, filename)
    extension = os.path.splitext(filename)[1]

    if extension.lower() == ".tmp":
        if os.path.exists(file_path):
            os.remove(file_path)
            return f"File '{filename}' removed successfully."
        else:
            return f"File '{filename}' not present."
    else:
        return f"File '{filename}' is not a tmp file."

@mcp.prompt()
def useful_helper_prompt(lang: str) -> str:
    return f"You are an expert in {lang}. create or update file using {lang} code."

if __name__ == "__main__":
    mcp.run(transport="http", host="0.0.0.0", port=8001, log_level="DEBUG")

Source: MCP Sensor Observability

---

MCP File Server (STDIO) — Python

from mcp.server.fastmcp import FastMCP
import os

mcp = FastMCP("FileServer")
WORKING_DIR = os.path.abspath("./mcp-data")

os.makedirs(WORKING_DIR, exist_ok=True)

@mcp.resource("config://app-settings")
def get_config() -> str:
    return "Mode: Development; Version: 1.0.0; Allowed-Extensions: .txt, .md"

@mcp.resource("folder://explorer")
def list_files() -> str:
    files = os.listdir(WORKING_DIR)
    return "\n".join(files) if files else "The directory is empty."

@mcp.tool()
def write_file(filename: str, content: str) -> str:
    path = os.path.join(WORKING_DIR, filename)
    with open(path, "w") as f:
        f.write(content)
    return f"File '{filename}' written successfully."

@mcp.tool()
def remove_file(filename: str) -> str:
    file_path = os.path.join(WORKING_DIR, filename)
    extension = os.path.splitext(filename)[1]

    if extension.lower() == ".tmp":
        if os.path.exists(file_path):
            os.remove(file_path)
            return f"File '{filename}' removed successfully."
        else:
            return f"File '{filename}' not present."
    else:
        return f"File '{filename}' is not a tmp file."

@mcp.prompt()
def useful_helper_prompt(lang: str) -> str:
    return f"You are an expert in {lang}. create or update file using {lang} code."

if __name__ == "__main__":
    mcp.run()

Source: MCP Sensor Observability

MCP Client Script — JavaScript

#!/usr/bin/env node

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
import yargs from "yargs";
import { hideBin } from "yargs/helpers";

async function main() {
  const argv = yargs(hideBin(process.argv))
    .option("mcp_server", {
      type: "string",
      demandOption: true,
      describe: "URL (http/sse) or local server script path (stdio)",
    })
    .option("mcp_auth_token", {
      type: "string",
      describe: "Optional bearer token for authentication",
    })
    .command("tools", "Tool operations", (yargs) => {
      yargs
        .option("list", { type: "boolean", describe: "List available tools" })
        .option("tool_cmd", { type: "string", describe: "Name of the tool to call" })
        .option("tool_args", { type: "string", default: "{}", describe: "JSON string of tool arguments" });
    })
    .command("resources", "Resource operations", (yargs) => {
      yargs
        .option("list", { type: "boolean", describe: "List available resources" })
        .option("resource_uri", { type: "string", describe: "URI of the resource to read" });
    })
    .command("prompts", "Prompt operations", (yargs) => {
      yargs
        .option("list", { type: "boolean", describe: "List available prompts" })
        .option("prompt_name", { type: "string", describe: "Name of the prompt to fetch" })
        .option("prompt_args", { type: "string", default: "{}", describe: "JSON string of prompt arguments" });
    })
    .demandCommand(1, "You must specify an action: tools, resources, or prompts")
    .help()
    .argv;

  const actionCmd = argv._[0];
  const mcpServer = argv.mcp_server;
  const authToken = argv.mcp_auth_token;

  let transport;
  if (mcpServer.startsWith("http://") || mcpServer.startsWith("https://")) {
    const url = new URL(mcpServer);
    const authProvider = authToken
      ? {
          async tokens() {
            return { access_token: authToken, token_type: "Bearer" };
          },
        }
      : undefined;

    if (mcpServer.includes("/sse")) {
      transport = new SSEClientTransport(url, { authProvider });
    } else {
      transport = new StreamableHTTPClientTransport(url, { authProvider });
    }
  } else {
    const command = mcpServer.endsWith(".js") ? "node" : "python";
    transport = new StdioClientTransport({ command, args: [mcpServer] });
  }

  const client = new Client(
    { name: "mcp-client", version: "1.0.0" },
    { capabilities: {} }
  );

  await client.connect(transport);

  try {
    if (actionCmd === "tools") {
      if (argv.list) {
        const result = await client.listTools();
        console.log("Available Tools:", JSON.stringify(result.tools, null, 2));
      } else if (argv.tool_cmd) {
        const toolArgs = JSON.parse(argv.tool_args);
        const result = await client.callTool({ name: argv.tool_cmd, arguments: toolArgs });
        console.log("Tool Result:", JSON.stringify(result, null, 2));
      }

    } else if (actionCmd === "resources") {
      if (argv.list) {
        const result = await client.listResources();
        console.log("Available Resources:", JSON.stringify(result.resources, null, 2));
      } else if (argv.resource_uri) {
        const result = await client.readResource({ uri: argv.resource_uri });
        console.log("Resource Content:", JSON.stringify(result, null, 2));
      }

    } else if (actionCmd === "prompts") {
      if (argv.list) {
        const result = await client.listPrompts();
        console.log("Available Prompts:", JSON.stringify(result.prompts, null, 2));
      } else if (argv.prompt_name) {
        const promptArgs = JSON.parse(argv.prompt_args);
        const result = await client.getPrompt({ name: argv.prompt_name, arguments: promptArgs });
        console.log("Prompt Output:", JSON.stringify(result, null, 2));
      }
    }
  } finally {
    await client.close();
  }
}

main().catch(console.error);

MCP File Server — JavaScript

import express from "express";
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
import { z } from "zod";
import fs from "fs/promises";
import fsSync from "fs";
import path from "path";

const WORKING_DIR = path.resolve("./mcp-data");

fsSync.mkdirSync(WORKING_DIR, { recursive: true });

const validTokens = {
  "dev-token": { client_id: "dev-client", scopes: ["read", "write"] },
};
const requiredScope = "read";

function createServer() {
  const server = new McpServer({ name: "FileServer", version: "1.0.0" });

  server.resource(
    "app-settings",
    "config://app-settings",
    async () => ({
      contents: [
        {
          uri: "config://app-settings",
          mimeType: "text/plain",
          text: "Mode: Development; Version: 1.0.0; Allowed-Extensions: .txt, .md",
        },
      ],
    })
  );

  server.resource(
    "explorer",
    "folder://explorer",
    async () => {
      const files = await fs.readdir(WORKING_DIR);
      const text = files.length > 0 ? files.join("\n") : "The directory is empty.";
      return {
        contents: [{ uri: "folder://explorer", mimeType: "text/plain", text }],
      };
    }
  );

  server.tool(
    "write_file",
    {
      filename: z.string().describe("Name of the file to write"),
      content: z.string().describe("Content to write into the file"),
    },
    async ({ filename, content }) => {
      const filePath = path.join(WORKING_DIR, filename);
      await fs.writeFile(filePath, content, "utf8");
      return {
        content: [{ type: "text", text: `File '${filename}' written successfully.` }],
      };
    }
  );

  server.tool(
    "read_file",
    {
      filename: z.string().describe("Name of the file to read"),
    },
    async ({ filename }) => {
      const filePath = path.join(WORKING_DIR, filename);
      try {
        const content = await fs.readFile(filePath, "utf8");
        return { content: [{ type: "text", text: content }] };
      } catch {
        return { content: [{ type: "text", text: `File ${filename} not found!!` }] };
      }
    }
  );

  server.tool(
    "remove_file",
    {
      filename: z.string().describe("Name of the .tmp file to remove"),
    },
    async ({ filename }) => {
      const ext = path.extname(filename).toLowerCase();
      const filePath = path.join(WORKING_DIR, filename);

      let status;
      if (ext !== ".tmp") {
        status = `File '${filename}' is not a tmp file.`;
      } else {
        try {
          await fs.access(filePath);
          await fs.unlink(filePath);
          status = `File '${filename}' removed successfully.`;
        } catch {
          status = `File '${filename}' not present.`;
        }
      }

      return { content: [{ type: "text", text: status }] };
    }
  );

  server.prompt(
    "useful_helper_prompt",
    { lang: z.string() },
    ({ lang }) => ({
      messages: [
        {
          role: "user",
          content: {
            type: "text",
            text: `You are an expert in ${lang}.  create or update file using ${lang} code.`,
          },
        },
      ],
    })
  );

  return server;
}

const app = express();
app.use(express.json());

const authMiddleware = (req, res, next) => {
  console.log(`[${new Date().toISOString()}] Incoming ${req.method} request to ${req.path}`);
  const token = req.headers["authorization"]?.split(" ")[1];
  const entry = token ? validTokens[token] : null;

  if (!entry || !entry.scopes.includes(requiredScope)) {
    console.warn("Auth Failed");
    return res.status(403).json({ error: "Forbidden" });
  }

  console.log(`Auth OK for ${entry.client_id}`);
  next();
};

app.post("/mcp", authMiddleware, async (req, res) => {
  const server = createServer();
  try {
    const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined });
    await server.connect(transport);
    await transport.handleRequest(req, res, req.body);
  } catch (error) {
    console.error("Request Error:", error);
    res.status(500).json({ error: "Internal Server Error" });
  }
});

const PORT = 8001;
app.listen(PORT, "0.0.0.0", () => {
  console.log(`FileServer active at http://0.0.0.0:${PORT}/mcp  |  Auth: Bearer dev-token`);
});

MCP File Server (STDIO) — JavaScript

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import fs from "fs/promises";
import fsSync from "fs";
import path from "path";

const WORKING_DIR = path.resolve("./mcp-data");

fsSync.mkdirSync(WORKING_DIR, { recursive: true });

const server = new McpServer({ name: "FileAgentServer", version: "1.0.0" });

server.resource(
  "app-settings",
  "config://app-settings",
  async () => ({
    contents: [
      {
        uri: "config://app-settings",
        mimeType: "text/plain",
        text: "Mode: Development; Version: 1.0.0; Allowed-Extensions: .txt, .md",
      },
    ],
  })
);

server.resource(
  "explorer",
  "folder://explorer",
  async () => {
    const files = await fs.readdir(WORKING_DIR);
    const text = files.length > 0 ? files.join("\n") : "The directory is empty.";
    return {
      contents: [{ uri: "folder://explorer", mimeType: "text/plain", text }],
    };
  }
);

server.tool(
  "write_file",
  {
    filename: z.string().describe("Name of the file to write"),
    content: z.string().describe("Content to write into the file"),
  },
  async ({ filename, content }) => {
    const filePath = path.join(WORKING_DIR, filename);
    await fs.writeFile(filePath, content, "utf8");
    return {
      content: [{ type: "text", text: `File '${filename}' written successfully.` }],
    };
  }
);

server.tool(
  "remove_file",
  {
    filename: z.string().describe("Name of the .tmp file to remove"),
  },
  async ({ filename }) => {
    const ext = path.extname(filename).toLowerCase();
    const filePath = path.join(WORKING_DIR, filename);

    let status;
    if (ext !== ".tmp") {
      status = `File '${filename}' is not a tmp file.`;
    } else {
      try {
        await fs.access(filePath);
        await fs.unlink(filePath);
        status = `File '${filename}' removed successfully.`;
      } catch {
        status = `File '${filename}' not present.`;
      }
    }

    return {
      content: [{ type: "text", text: status }],
    };
  }
);

const transport = new StdioServerTransport();
await server.connect(transport);

package.json - Javascript

{
  "type": "module",
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.7.0",
    "bluejs": "file:..",
    "eventsource": "^4.1.0",
    "express": "^4.18.2",
    "zod": "^3.22.4",
    "glob": "^10.3.10",
    "yargs": "^17.7.2"
  }
}