tool-development-guide

This guide outlines the core coding patterns for implementing tools in the Atlan MCP server.

Tool Implementation Pattern

1. Define core function in tools.py
2. Register function as MCP tool in server.py
3. Use appropriate type hints for PyAtlan and MCP compatibility

Core Function Template (tools.py)

def implement_atlan_tool(
    param1: str,
    param2: Optional[int] = None,
    param3: Optional[Dict[str, Any]] = None
) -> Any:
    """
    Implement an Atlan tool with appropriate parameters.
    """
    logger.info(f"Starting tool execution with parameters: param1={param1}")

    try:
        # Tool-specific implementation
        result = execute_atlan_operation(param1, param2, param3)

        logger.info(f"Tool execution completed successfully")
        return result
    except Exception as e:
        logger.error(f"Error executing tool: {str(e)}")
        logger.exception("Exception details:")
        return get_appropriate_default_value()

MCP Tool Registration Template (server.py)

@mcp.tool()
def registered_tool_name(
    param1: str,
    param2: Optional[int] = None,
    param3: Optional[Dict[str, Any]] = None
) -> Any:
    """
    Tool description with clear purpose.

    Args:
        param1: First parameter description
        param2: Second parameter description
        param3: Third parameter description (complex structure)

    Returns:
        Description of return value

    Example:
        registered_tool_name("value1", 42, {"key": "value"})
    """
    return implement_atlan_tool(param1, param2, param3)

Common Operation Patterns

Asset Search Operation

def search_operation(criteria):
    search = FluentSearch()
    # Add filters based on criteria
    request = search.to_request()
    results = list(atlan_client.asset.search(request).current_page())
    return results

Asset Retrieval Operation

def get_asset_operation(qualified_name, asset_type):
    try:
        asset = asset_type.get_by_qualified_name(
            qualified_name=qualified_name,
            min_ext_info=True,
            atlan_client=atlan_client
        )
        return asset
    except Exception as e:
        logger.error(f"Failed to get asset: {e}")
        return None