How to Extend the MCP Ambari API Server with New Ambari Endpoints

Adding new Ambari endpoints requires creating an async function decorated with @mcp.tool() and @log_tool in src/mcp_ambari_api/mcp_main.py, utilizing the existing make_ambari_request helper for HTTP operations.

The call518/mcp-ambari-api repository provides a FastMCP-based integration layer for Apache Ambari, exposing cluster operations as tools for LLM clients. To extend the server with new Ambari API endpoints, you follow a repeatable pattern that leverages centralized utilities for authentication, logging, and error handling.

Understanding the Extension Architecture

The server architecture centers on a single FastMCP instance that automatically registers decorated functions as protocol tools.

The FastMCP Instance and Tool Registration

In src/mcp_ambari_api/mcp_main.py (approximately line 79), the FastMCP instance initializes once at import time:

mcp = FastMCP("mcp-ambari-api")

Any function decorated with @mcp.tool() becomes immediately discoverable via the MCP protocol without additional registration code. The @log_tool decorator (from src/mcp_ambari_api/functions.py) wraps each call to provide uniform timing, argument logging, and error tracking.

Centralized Helper Functions

The src/mcp_ambari_api/functions.py module provides reusable utilities:

• make_ambari_request: Handles HTTP authentication, JSON parsing, and error normalization for all Ambari REST calls.
• log_tool: Records start times, arguments, duration, and success/failure states.
• AMBARI_CLUSTER_NAME: A constant importing the configured cluster identifier.

Reusing these helpers ensures consistent error handling and logging across all endpoints.

Step-by-Step Guide to Adding a New Endpoint

Follow this sequence to integrate any Ambari REST operation:

1. Import utilities at the top of src/mcp_ambari_api/mcp_main.py:

   from mcp_ambari_api.functions import (
       make_ambari_request,
       log_tool,
       AMBARI_CLUSTER_NAME,
   )

2. Define an async function that constructs the Ambari URL, invokes make_ambari_request, and formats the response into a user-friendly string.

3. Apply decorators in this order:

   • @mcp.tool(title="Your Tool Name") (exposes to MCP protocol)
   • @log_tool (adds observability)

4. Add a docstring with a clear first line—FastMCP uses this as the tool's help text.

5. Test locally using the MCP inspector or streamable-HTTP mode.

6. Deploy by committing the changes; the FastMCP instance detects new tools automatically at startup.

Code Implementation Examples

Creating a Cluster Topology Tool

This example retrieves host and service information by querying /clusters/{cluster} with specific fields:

@mcp.tool(title="Cluster Topology")
@log_tool
async def get_cluster_topology() -> str:
    """
    Retrieve a compact view of the cluster's topology:
    hosts → components → installed services.
    """
    cluster = AMBARI_CLUSTER_NAME
    try:
        endpoint = f"/clusters/{cluster}?fields=Clusters/hosts,Clusters/services"
        data = await make_ambari_request(endpoint)

        if data.get("error"):
            return f"Error: {data['error']}"

        hosts = data.get("Clusters", {}).get("hosts", [])
        services = data.get("Clusters", {}).get("services", [])

        lines = [
            f"Cluster Topology for '{cluster}':",
            "=" * 40,
            f"Hosts ({len(hosts)}):",
        ]
        for h in hosts:
            host_name = h.get("Hosts", {}).get("host_name", "unknown")
            lines.append(f"  • {host_name}")

        lines.append("")
        lines.append(f"Services ({len(services)}):")
        for s in services:
            svc = s.get("ServiceInfo", {})
            lines.append(f"  • {svc.get('service_name', 'unknown')} [{svc.get('state', 'UNKNOWN')}]")

        return "\n".join(lines)

    except Exception as exc:
        return f"Error: Exception while fetching topology – {exc}"

Adding a Resource Endpoint

For hierarchical data that accepts parameters, use @mcp.resource() instead of @mcp.tool():

@mcp.resource(path="/hosts/{host_name}/metrics")
@log_tool
async def host_metrics_resource(host_name: str, metric: str = "cpu_user") -> str:
    """
    Return a single metric value for a given host.
    """
    hostnames = host_name
    series = await fetch_metric_series(metric, hostnames=hostnames, duration_ms=5*60*1000)
    if not series:
        return f"No data for metric '{metric}' on host '{host_name}'."
    latest = series[-1]["value"]
    return f"{host_name} – {metric}: {latest}"

Resources are addressable via MCP paths like /hosts/<hostname>/metrics and accept arguments directly from LLM clients.

Testing Your New Endpoint

Start the server in streamable-HTTP mode to verify functionality:

PYTHONPATH=./src uv run python -m mcp_ambari_api --type streamable-http --host 0.0.0.0 --port 8000

Query the endpoint using curl or the MCP inspector:

curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{"method":"get_cluster_topology","params":[]}'

The response returns the formatted string defined in your function. Check server logs (via the log_tool decorator) to verify execution timing and parameter handling.

Summary

• Primary file: Add all new endpoints to src/mcp_ambari_api/mcp_main.py following the existing pattern.
• Required decorators: Use @mcp.tool() for RPC-style operations or @mcp.resource() for REST-like paths, always combined with @log_tool.
• HTTP handling: Delegate all Ambari REST calls to make_ambari_request from src/mcp_ambari_api/functions.py to maintain consistent authentication and error handling.
• Automatic registration: The FastMCP instance registers tools at import time, requiring no manual registry updates.
• Testing: Use uv run python -m mcp_ambari_api with --type streamable-http for local validation before deployment.

Frequently Asked Questions

What file should I edit to add new Ambari endpoints?

Edit src/mcp_ambari_api/mcp_main.py. This file contains the FastMCP instance and serves as the central registry for all tools and resources. Import helper functions from src/mcp_ambari_api/functions.py at the top of this file.

How does the @mcp.tool() decorator work?

The @mcp.tool() decorator registers the async function as an RPC method accessible through the MCP protocol. FastMCP uses the function name as the method identifier and the first line of the docstring as the description. When combined with @log_tool, it automatically captures execution metrics and error states.

Can I add endpoints that accept parameters?

Yes. Define parameters in the async function signature—standard Python type hints map directly to MCP schema definitions. For parameterized resources (like host-specific metrics), use @mcp.resource() with path templates such as /hosts/{host_name}/metrics to create REST-like endpoints that accept dynamic arguments.

How do I test new endpoints locally?

Run the server with uv run python -m mcp_ambari_api --type streamable-http --host 0.0.0.0 --port 8000 from the repository root, then invoke the tool via HTTP POST to http://localhost:8000/mcp with a JSON payload containing the method name and parameters. Alternatively, use the MCP Inspector for interactive testing and debugging.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}