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

> Extend the MCP Ambari API server by adding new Ambari endpoints. Learn to create async functions using decorators and the make ambari request helper.

- Repository: [JungJungIn/mcp-ambari-api](https://github.com/call518/mcp-ambari-api)
- Tags: how-to-guide
- Published: 2026-02-26

---

**Adding new Ambari endpoints requires creating an async function decorated with `@mcp.tool()` and `@log_tool` in [`src/mcp_ambari_api/mcp_main.py`](https://github.com/call518/mcp-ambari-api/blob/main/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`](https://github.com/call518/mcp-ambari-api/blob/main/src/mcp_ambari_api/mcp_main.py) (approximately line 79), the FastMCP instance initializes once at import time:

```python
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`](https://github.com/call518/mcp-ambari-api/blob/main/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`](https://github.com/call518/mcp-ambari-api/blob/main/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`](https://github.com/call518/mcp-ambari-api/blob/main/src/mcp_ambari_api/mcp_main.py):
   ```python
   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:

```python
@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()`:

```python
@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:

```bash
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:

```bash
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`](https://github.com/call518/mcp-ambari-api/blob/main/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`](https://github.com/call518/mcp-ambari-api/blob/main/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`](https://github.com/call518/mcp-ambari-api/blob/main/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`](https://github.com/call518/mcp-ambari-api/blob/main/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.