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"
  }
}

chat_agent.py

import os
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import chain


template = ChatPromptTemplate.from_messages([('system', 'You are a helpful assistant'),
                                             ('human', '{question}')])

llm = ChatOpenAI(base_url="<MASKED_BASE_URL>",
                 model="<MASKED_MODEL_NAME>",
                 temperature=0)


@chain
def chatbot(values):
    prompt = template.invoke(values)
    return llm.invoke(prompt)
    
for chunk in chatbot.stream({"question": "why is sky blue?"}):
    print(chunk.content)

help_agent_crewai.py

help_agent_crewai.py

import json
from crewai import LLM, Agent, Crew, Task, Process

class Assistant:
    def help_agent(self) -> Agent:
        return Agent(
            role="Helpful assistant",
            goal="Helpful assistant. answer user queries",
            backstory="A helpful assistant. can answer any topic",
            llm=LLM(
                model="<MASKED_MODEL_NAME>",
                base_url="<MASKED_BASE_URL>",
                stream=True
            ),
            verbose=True
        )

    def task_one(self, agent: Agent) -> Task:
        return Task(
            description="why sky is blue?",
            expected_output="provide more scientic answer",
            agent=agent # Uses the shared agent instance
        )

    def crew(self) -> Crew:
        # Create a single instance of the agent to share across Task and Crew
        shared_agent = self.help_agent()
        return Crew(
            agents=[shared_agent],
            tasks=[self.task_one(shared_agent)],
            process=Process.sequential,
            verbose=True
        )

try:
    crew_output = Assistant().crew().kickoff()
    print(f"Raw Output: {crew_output.raw}")
    if crew_output.json_dict:
        print(f"JSON Output: {json.dumps(crew_output.json_dict, indent=2)}")
    if crew_output.pydantic:
        print(f"Pydantic Output: {crew_output.pydantic}")
    print(f"Tasks Output: {crew_output.tasks_output}")
    print(f"Token Usage: {crew_output.token_usage}")
except Exception as err:
    print(f"An error occurred: {err}")

search_agent.py

dummy_search_agent.py

import os
from anthropic import Anthropic 


try:
    client = Anthropic(
            api_key=os.environ.get("<MASKED_API_KEY_ENV_VAR>"),
            base_url="<MASKED_BASE_URL>")

    print("Agent is running!", flush=True)
    message = client.messages.create(
        model="<MASKED_MODEL_NAME>",
        max_tokens=1024,
        messages=[
            {"role": "user", "content": "Hello, Claude. Why is the sky so blue?"}
        ]
    )

    print(message.content[0].text)
    print("execution completed successfully", flush=True)

except Exception as err:
    print(f"An error occurred: {err}")

search_agent_gemini.py

search_agent_gemini.py

import asyncio
from google.adk.agents import Agent
from google.adk.tools import google_search
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

QUERY = "why is the sky blue?"

agent = Agent(
    name="search_agent",
    model="<MASKED_MODEL_NAME>",
    instruction="Helpful assistant. Answer user queries using web search.",
    tools=[google_search],
)

async def main():
    try:
        session_service = InMemorySessionService()
        session = await session_service.create_session(
            app_name="<MASKED_APP_NAME>", 
            user_id="<MASKED_USER_ID>"
        )

        runner = Runner(
            agent=agent, 
            app_name="<MASKED_APP_NAME>", 
            session_service=session_service
        )

        message = types.Content(role="user", parts=[types.Part(text=QUERY)])  
        print("Agent is running!", flush=True)
        
        async for event in runner.run_async(
            user_id="<MASKED_USER_ID>", 
            session_id=session.id, 
            new_message=message
        ):
            if event.is_final_response() and event.content:
                print(event.content.parts[0].text)
        print("execution completed successfully", flush=True)
    except Exception as err:
        print(f"An error occurred: {err}")

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

hello_world_oai.py

hello_world_oai.py

import os
from openai import OpenAI

try:
    client = OpenAI(
        # Defaults to the standard environment variable
        api_key=os.environ.get("<MASKED_API_KEY_ENV_VAR>"),
        base_url="<MASKED_BASE_URL>"
    )
    print("Agent is running!", flush=True)
    response = client.chat.completions.create(
        model="<MASKED_MODEL_NAME>",
        messages=[{"role": "user", "content": "Hello world, using the Python SDK!"}],
    )

    print(response.choices[0].message.content)
    print("execution completed successfully", flush=True)
except Exception as err:
    print(f"An error occurred: {err}")

search_agent_litellm.py

search_agent_litellm.py

import os
from litellm import completion

# Ensure the API key environment variable is set
# It can be set in the terminal or within the script like this:
# os.environ["<MASKED_API_KEY_ENV_VAR>"] = "api-key-here"

messages = [{
    "content": "Hello, what is a rainbow?",
    "role": "user"
}]

# Call the completion function
# To use a different provider, change the 'model' parameter to the provider's format 
try:
    print("Agent is running!", flush=True)
    response = completion(
        api_base="<MASKED_BASE_URL>",
        api_key=os.environ.get("<MASKED_API_KEY_ENV_VAR>"),
        model="<MASKED_MODEL_NAME>",
        messages=messages
    )

    # Print the response content
    print(response.choices[0].message.content)
    print("execution completed successfully", flush=True)
except Exception as err:
    print(f"An error occurred: {err}")